diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index 35f0b06eadc..efc20044333 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/packages/core/src/session/runner/retry.ts @@ -48,10 +48,12 @@ export const schedule = ( assistantMessageID: () => SessionMessage.ID, ) => Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe( + Schedule.jittered, Schedule.setInputType(), Schedule.modifyDelay(({ input: failure, duration: delay }) => { const minimum = retryAfter(failure) - return Effect.succeed(minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum))) + const duration = minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum)) + return Effect.succeed(Duration.millis(Math.ceil(Duration.toMillis(duration)))) }), Schedule.tap((metadata) => bus.publish(SessionEvent.RetryScheduled, { diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 5bfbedf4d6f..40efdbb4bb0 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -3951,7 +3951,7 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("retries eligible pre-output failures after exponential backoff", () => + it.effect("bounds jittered exponential backoff for eligible pre-output failures", () => Effect.gen(function* () { const session = yield* setup yield* admit(session, "Retry transport") @@ -3960,9 +3960,9 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - yield* TestClock.adjust("1999 millis") + yield* TestClock.adjust("1599 millis") expect(requests).toHaveLength(1) - yield* TestClock.adjust("1 millis") + yield* TestClock.adjust("801 millis") yield* Fiber.join(run) expect(requests).toHaveLength(2) @@ -3987,7 +3987,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(requests).toHaveLength(2) @@ -4032,7 +4032,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(requests).toHaveLength(2) @@ -4089,7 +4089,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(requests[1]?.messages.at(-2)).toMatchObject({ @@ -4130,7 +4130,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(executions).toEqual(["settled"]) @@ -4169,7 +4169,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(messageRoles(requests[1])).toEqual(["user", "assistant", "tool", "user"]) @@ -4207,7 +4207,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - for (const [index, delay] of [2_000, 4_000, 8_000, 16_000].entries()) { + for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) { yield* TestClock.adjust(delay) yield* TestLLM.wait(index + 2) } @@ -4228,7 +4228,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - for (const [index, delay] of [2_000, 4_000, 8_000, 16_000].entries()) { + for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) { yield* TestClock.adjust(delay) yield* TestLLM.wait(index + 2) } @@ -4243,12 +4243,15 @@ describe("SessionRunnerLLM", () => { .orderBy(asc(EventTable.seq)) .all() .pipe(Effect.orDie) - expect(retries.map((event) => event.data)).toMatchObject([ - { attempt: 2, at: 2_000 }, - { attempt: 3, at: 6_000 }, - { attempt: 4, at: 14_000 }, - { attempt: 5, at: 30_000 }, - ]) + for (const [index, range] of [ + [1_600, 2_400], + [4_800, 7_200], + [11_200, 16_800], + [24_000, 36_000], + ].entries()) { + expect(retries[index]?.data.at).toBeGreaterThanOrEqual(range[0]!) + expect(retries[index]?.data.at).toBeLessThanOrEqual(range[1]!) + } expect((yield* recordedEventTypes(sessionID)).filter((type) => type === "session.step.started.1")).toHaveLength(5) const assistant = requireAssistant(yield* session.context(sessionID)) expect(yield* recordedStepSettlementEvents(sessionID, assistant.id)).toMatchObject([ @@ -4278,7 +4281,7 @@ describe("SessionRunnerLLM", () => { const run = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* TestLLM.wait(1) - yield* TestClock.adjust("2 seconds") + yield* TestClock.adjust("2400 millis") yield* Fiber.join(run) expect(requests).toHaveLength(3)