From a38d44265b687619109f91a17e1934e334612248 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 6 Aug 2026 00:04:27 -0500 Subject: [PATCH] fix(core): continue interrupted responses (#40576) --- packages/core/src/session/runner/llm.ts | 28 ++++ packages/core/test/session-runner.test.ts | 191 +++++++++++++++++++++- 2 files changed, 213 insertions(+), 6 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index e92685c68dd..04bf95ad07e 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -37,6 +37,7 @@ import { SessionUsage } from "../usage" type CallOutcome = Data.TaggedEnum<{ Completed: { readonly needsContinuation: boolean; readonly step: number } Retry: { readonly step: number } + Continue: { readonly cause: AIError; readonly error: SessionRunnerRetry.RetryableFailure["error"]; readonly step: number } Restart: { readonly step: number; readonly recoveredOverflow: boolean } }> const CallOutcome = Data.taggedEnum() @@ -91,6 +92,8 @@ const classifyToolExits = ( const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const +const CONTINUE_AFTER_INCOMPLETE_STREAM = + "The previous response was interrupted. Continue from where you left off without repeating completed content." const layer = Layer.effect( Service, @@ -187,6 +190,20 @@ const layer = Layer.effect( assistantMessageID, ).pipe(Effect.catchTag("SessionRunner.RetryableFailure", waitForRetry)) if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: outcome.step } + if (outcome._tag === "Continue") { + yield* retry( + new SessionRunnerRetry.RetryableFailure({ + cause: outcome.cause, + error: outcome.error, + step: outcome.step, + }), + ).pipe(Pull.catchDone(() => Effect.fail(outcome.cause))) + yield* bus.publish(SessionEvent.Synthetic, { + sessionID, + text: CONTINUE_AFTER_INCOMPLETE_STREAM, + }) + assistantMessageID = SessionMessage.ID.create() + } if (outcome._tag === "Restart") { if (outcome.recoveredOverflow) recoverOverflow = false assistantMessageID = SessionMessage.ID.create() @@ -426,6 +443,17 @@ const layer = Layer.effect( }) } + const incompleteStream = + llmFailure?.reason._tag === "InvalidProviderOutput" && + llmFailure.reason.classification === "incomplete-stream" + const toolsAllowContinuation = tools.declines.length === 0 && !tools.interrupted + if (llmError && incompleteStream && record.outputStarted && toolsAllowContinuation) + return CallOutcome.Continue({ + cause: llmFailure, + error: llmError, + step: currentStep, + }) + if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) if (tools.declines.length > 0) return yield* Effect.interrupt if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 6092c10edfe..426a8cd4ffb 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -524,6 +524,9 @@ const incompleteStream = () => }), }) +const INCOMPLETE_STREAM_CONTINUATION = + "The previous response was interrupted. Continue from where you left off without repeating completed content." + const invalidRequest = () => new AIError({ module: "test", @@ -4007,10 +4010,11 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("does not retry eligible failures after observable output", () => + it.effect("continues an incomplete stream after observable text", () => Effect.gen(function* () { const session = yield* setup const failure = incompleteStream() + yield* admit(session, "Continue partial output") yield* TestLLM.push( TestLLM.failAfter( failure, @@ -4019,19 +4023,194 @@ describe("SessionRunnerLLM", () => { LLMEvent.textDelta({ id: "partial-rate-limit", text: "Partial" }), ), ) + yield* TestLLM.push(TestLLM.text(" continuation", "continued-text")) - expect(yield* runPrompt(session, "Do not replay partial output").pipe(Effect.flip)).toBe(failure) - expect(requests).toHaveLength(1) - expect(yield* recordedEventTypes(sessionID)).not.toContain("session.retry.scheduled.1") - expect(yield* session.context(sessionID)).toMatchObject([ - { type: "user" }, + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* TestLLM.wait(1) + yield* TestClock.adjust("2 seconds") + yield* Fiber.join(run) + + expect(requests).toHaveLength(2) + expect(requests[1]?.messages.at(-2)).toMatchObject({ + role: "assistant", + content: [{ type: "text", text: "Partial" }], + }) + expect(requests[1]?.messages.at(-1)).toMatchObject({ + role: "user", + content: [ + { + type: "text", + text: INCOMPLETE_STREAM_CONTINUATION, + }, + ], + }) + const context = yield* session.context(sessionID) + expect(context).toMatchObject([ + { type: "user", text: "Continue partial output" }, { type: "assistant", finish: "error", error: { type: "provider.invalid-output" }, content: [{ type: "text", text: "Partial" }], }, + { + type: "synthetic", + text: INCOMPLETE_STREAM_CONTINUATION, + }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: " continuation" }] }, ]) + const assistants = context.filter((message) => message.type === "assistant") + expect(new Set(assistants.map((message) => message.id)).size).toBe(2) + expect(context.find((message) => message.type === "synthetic")?.description).toBeUndefined() + expect(yield* recordedEventTypes(sessionID)).toContain("session.retry.scheduled.1") + yield* replaySessionProjection(sessionID) + expect(yield* session.context(sessionID)).toMatchObject(context) + }), + ) + + it.effect("lowers interrupted reasoning before continuing an incomplete stream", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Continue interrupted reasoning") + yield* TestLLM.push( + TestLLM.failAfter( + incompleteStream(), + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ id: "partial-reasoning" }), + LLMEvent.reasoningDelta({ id: "partial-reasoning", text: "Partial thought" }), + ), + ) + yield* TestLLM.push(TestLLM.text("Recovered", "reasoning-recovery")) + + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* TestLLM.wait(1) + yield* TestClock.adjust("2 seconds") + yield* Fiber.join(run) + + expect(requests[1]?.messages.at(-2)).toMatchObject({ + role: "assistant", + content: [{ type: "text", text: "Partial thought" }], + }) + expect(requests[1]?.messages.at(-1)).toMatchObject({ + role: "user", + content: [ + { + type: "text", + text: INCOMPLETE_STREAM_CONTINUATION, + }, + ], + }) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user" }, + { type: "assistant", finish: "error", content: [{ type: "reasoning", text: "Partial thought" }] }, + { type: "synthetic" }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] }, + ]) + }), + ) + + it.effect("continues an incomplete stream after settling a local tool", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Continue after tool") + yield* TestLLM.push( + TestLLM.failAfter( + incompleteStream(), + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call-before-close", name: "echo", input: { text: "settled" } }), + ), + ) + yield* TestLLM.push(TestLLM.text("Recovered", "tool-recovery")) + + 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* Fiber.join(run) + + expect(executions).toEqual(["settled"]) + expect(requests[1]?.messages.slice(-3)).toMatchObject([ + { + role: "assistant", + content: [{ type: "tool-call", id: "call-before-close", name: "echo", input: { text: "settled" } }], + }, + { role: "tool", content: [{ type: "tool-result", id: "call-before-close" }] }, + { + role: "user", + content: [ + { + type: "text", + text: INCOMPLETE_STREAM_CONTINUATION, + }, + ], + }, + ]) + }), + ) + + it.effect("continues an incomplete stream after settling a local tool defect", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Continue after tool defect") + yield* TestLLM.push( + TestLLM.failAfter( + incompleteStream(), + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call-defect-before-close", name: "defect", input: {} }), + ), + ) + yield* TestLLM.push(TestLLM.text("Recovered", "tool-defect-recovery")) + + 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* Fiber.join(run) + + expect(messageRoles(requests[1])).toEqual(["user", "assistant", "tool", "user"]) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user" }, + { + type: "assistant", + content: [ + { + type: "tool", + id: "call-defect-before-close", + state: { status: "error", error: { type: "unknown", message: "unexpected tool defect" } }, + }, + ], + }, + { type: "synthetic", text: INCOMPLETE_STREAM_CONTINUATION }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] }, + ]) + }), + ) + + it.effect("stops incomplete stream continuations after five total attempts", () => + Effect.gen(function* () { + const session = yield* setup + yield* admit(session, "Exhaust partial continuations") + const failure = incompleteStream() + yield* TestLLM.always( + TestLLM.failAfter( + failure, + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "partial-exhaustion" }), + LLMEvent.textDelta({ id: "partial-exhaustion", text: "Partial" }), + ), + ) + + 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()) { + yield* TestClock.adjust(delay) + yield* TestLLM.wait(index + 2) + } + expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure) + expect(requests).toHaveLength(5) + const context = yield* session.context(sessionID) + expect(context.filter((message) => message.type === "assistant")).toHaveLength(5) + expect(context.filter((message) => message.type === "synthetic")).toHaveLength(4) }), )