fix(core): continue after settled tool errors

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<CallOutcome>()
@@ -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)

packages/core/test/session-runner.test.ts

@@ -523,6 +523,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",
@@ -3996,10 +3999,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,
@@ -4008,19 +4012,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)
     }),
   )