fix(cli): retry empty incomplete responses

.changeset/retry-incomplete-responses.md

@@ -0,0 +1,5 @@
+---
+"@kilocode/cli": patch
+---
+
+Retry incomplete model responses that end without final output or tool activity while preserving partial answers and completed tools.

packages/opencode/src/kilocode/session/processor.ts

@@ -7,9 +7,12 @@ import { MessageV2 } from "@/session/message-v2"
 import { isRecord } from "@/util/record"
 import { parseReviewCommand, reviewCommandName } from "@/kilocode/review/command"
 import * as Log from "@opencode-ai/core/util/log"
-import { Effect } from "effect"
+import { Cause, Effect, Exit } from "effect"
 import { Flag } from "@opencode-ai/core/flag/flag"
 import { EffectBridge } from "@/effect/bridge"
+import type { LLMEvent, Usage } from "@opencode-ai/llm"
+import type { ProviderV2 } from "@opencode-ai/core/provider"
+import { SessionRetry } from "@/session/retry"
 
 export type ReviewTelemetry = {
   mode: "review"
@@ -20,6 +23,23 @@ export type ReviewTelemetry = {
 
 export namespace KiloSessionProcessor {
   const log = Log.create({ service: "session.processor.kilo" })
+  export const INCOMPLETE_RESPONSE_RETRIES = 2
+  export const INCOMPLETE_RESPONSE_MESSAGE =
+    "The provider repeatedly ended the response before returning usable output."
+  export class IncompleteResponseError extends Error {
+    constructor() {
+      super(INCOMPLETE_RESPONSE_MESSAGE)
+      this.name = "IncompleteResponseError"
+    }
+  }
+  export type Attempt = {
+    text: boolean
+    reasoning: boolean
+    tool: boolean
+    usage: boolean
+    finished: boolean
+    finish?: string
+  }
   export const OUTPUT_LENGTH_WARNING = "The model hit its output limit, so this response may be incomplete."
   export const REASONING_LENGTH_WARNING =
     "The model hit its output limit while reasoning and produced no actionable output. Try disabling reasoning or increasing the output limit."
@@ -172,9 +192,11 @@ export namespace KiloSessionProcessor {
     sessionID: SessionID
     abort: AbortSignal
     set: (sessionID: SessionID, status: SessionStatus.Info) => Effect.Effect<void>
+    used?: number
   }) {
+    const limit = Flag.KILO_SESSION_RETRY_LIMIT
     return {
-      limit: Flag.KILO_SESSION_RETRY_LIMIT,
+      limit: limit === undefined ? undefined : Math.max(0, limit - (input.used ?? 0)),
       offline: (info: { error: unknown; message: string }) =>
         handleOffline({
           error: info.error,
@@ -185,6 +207,85 @@ export namespace KiloSessionProcessor {
     }
   }
 
+  export function hasUsage(usage: Usage | undefined) {
+    if (!usage) return false
+    return [
+      usage.inputTokens,
+      usage.outputTokens,
+      usage.nonCachedInputTokens,
+      usage.cacheReadInputTokens,
+      usage.cacheWriteInputTokens,
+      usage.reasoningTokens,
+      usage.totalTokens,
+    ].some((value) => value !== undefined && value !== 0)
+  }
+
+  export function attempt(): Attempt {
+    return { text: false, reasoning: false, tool: false, usage: false, finished: false }
+  }
+
+  export function observe(attempt: Attempt, event: LLMEvent) {
+    if (event.type === "text-delta" && event.text.trim()) attempt.text = true
+    if (event.type === "reasoning-delta" && event.text.trim()) attempt.reasoning = true
+    if (event.type === "tool-call" || event.type === "tool-result" || event.type === "tool-error") attempt.tool = true
+    if (event.type === "step-finish") {
+      attempt.finished = true
+      attempt.finish = event.reason
+      attempt.usage ||= hasUsage(event.usage)
+    }
+    if (event.type === "finish" && !attempt.finished) {
+      attempt.finish = event.reason
+      attempt.usage ||= hasUsage(event.usage)
+    }
+  }
+
+  export function replayable(input: {
+    finish?: string
+    text: boolean
+    reasoning: boolean
+    tool: boolean
+    usage: boolean
+  }) {
+    if (input.finish !== undefined && input.finish !== "unknown") return false
+    return !input.text && !input.reasoning && !input.tool && !input.usage
+  }
+
+  export function blockRetry(error: ReturnType<typeof MessageV2.fromError>) {
+    const message = MessageV2.APIError.isInstance(error) ? error.data.message : "Response interrupted after output"
+    return new MessageV2.APIError({ message, isRetryable: false }).toObject()
+  }
+
+  export function recover(input: {
+    run: () => Effect.Effect<void, unknown>
+    replayable: () => boolean
+    discard: () => Effect.Effect<void>
+    set: (info: { attempt: number; message: string; next: number }) => Effect.Effect<void>
+  }) {
+    return Effect.gen(function* () {
+      for (const index of [0, 1, 2]) {
+        const result = yield* input.run().pipe(Effect.exit)
+        if (Exit.isFailure(result)) {
+          const error = Cause.squash(result.cause)
+          if (!(error instanceof IncompleteResponseError)) return yield* Effect.fail(error)
+        } else if (!input.replayable()) return
+
+        yield* input.discard()
+        if (index === INCOMPLETE_RESPONSE_RETRIES) return yield* Effect.fail(new IncompleteResponseError())
+        const wait = SessionRetry.delay(index + 1)
+        yield* input.set({ attempt: index + 1, message: INCOMPLETE_RESPONSE_MESSAGE, next: Date.now() + wait })
+        yield* Effect.sleep(`${wait} millis`)
+      }
+    })
+  }
+
+  export function parseError(error: unknown, input: { providerID: ProviderV2.ID; aborted: boolean }) {
+    if (!(error instanceof IncompleteResponseError)) return MessageV2.fromError(error, input)
+    return new MessageV2.APIError({
+      message: error.message,
+      isRetryable: true,
+    }).toObject()
+  }
+
   /**
    * Guard: if finish reason is "tool-calls" but no tool parts exist,
    * downgrade to "stop" to prevent an infinite loop (#7756).

packages/opencode/src/session/processor.ts

@@ -169,12 +169,21 @@ export const layer = Layer.effect(
       let aborted = false
       const ac = new AbortController() // kilocode_change — abort controller for offline handler
       const slog = log.clone().tag("session.id", input.sessionID).tag("messageID", input.assistantMessage.id)
+      let attempt = KiloSessionProcessor.attempt() // kilocode_change
 
+      // kilocode_change start
       const parse = (e: unknown) =>
-        MessageV2.fromError(e, {
+        KiloSessionProcessor.parseError(e, {
           providerID: input.model.providerID,
           aborted,
         })
+      const retryParse = (e: unknown) => {
+        const error = parse(e)
+        if (e instanceof KiloSessionProcessor.IncompleteResponseError) return KiloSessionProcessor.blockRetry(error)
+        if (attempt.text || attempt.reasoning || attempt.tool) return KiloSessionProcessor.blockRetry(error)
+        return error
+      }
+      // kilocode_change end
 
       const settleToolCall = Effect.fn("SessionProcessor.settleToolCall")(function* (toolCallID: string) {
         const done = ctx.toolcalls[toolCallID]?.done
@@ -466,10 +475,10 @@ export const layer = Layer.effect(
       }
 
       const handleEvent = Effect.fnUntraced(function* (value: StreamEvent) {
+        KiloSessionProcessor.observe(attempt, value) // kilocode_change
         switch (value.type) {
           case "reasoning-start":
             if (value.id in ctx.reasoningMap) return
-            ctx.step.reasoning = true // kilocode_change
             // TODO(v2): Temporary dual-write while migrating session messages to v2 events.
             if (mirrorAssistant) {
               yield* events.publish(SessionEvent.Reasoning.Started, {
@@ -496,6 +505,7 @@ export const layer = Layer.effect(
             // Match dev: silently drop orphan deltas (no preceding reasoning-start).
             if (!(value.id in ctx.reasoningMap)) return
             ctx.reasoningMap[value.id].text += value.text
+            if (value.text.trim()) ctx.step.reasoning = true // kilocode_change
             if (value.providerMetadata) ctx.reasoningMap[value.id].metadata = value.providerMetadata
             if (mirrorAssistant) {
               yield* events.publish(SessionEvent.Reasoning.Delta, {
@@ -526,7 +536,9 @@ export const layer = Layer.effect(
             if (ctx.assistantMessage.summary) {
               throw new Error(`Tool call not allowed while generating summary: ${value.name}`)
             }
-            ctx.step.tool = true // kilocode_change
+            // kilocode_change start
+            ctx.step.tool = true
+            // kilocode_change end
             yield* ensureToolCall(value)
             return
 
@@ -568,9 +580,7 @@ export const layer = Layer.effect(
             if (ctx.assistantMessage.summary) {
               throw new Error(`Tool call not allowed while generating summary: ${value.name}`)
             }
-            // kilocode_change start
-            ctx.step.tool = true
-            // kilocode_change end
+            ctx.step.tool = true // kilocode_change
             const toolCall = yield* ensureToolCall(value)
             const input = isRecord(value.input) ? value.input : { value: value.input }
             if (!toolCall.call.inputEnded) {
@@ -817,6 +827,19 @@ export const layer = Layer.effect(
             return
 
           case "step-finish": {
+            // kilocode_change start - retry only terminally incomplete attempts before settlement
+            if (
+              !mirrorAssistant &&
+              KiloSessionProcessor.replayable({
+                finish: attempt.finish,
+                text: attempt.text,
+                reasoning: attempt.reasoning,
+                tool: attempt.tool,
+                usage: attempt.usage,
+              })
+            )
+              return yield* Effect.fail(new KiloSessionProcessor.IncompleteResponseError())
+            // kilocode_change end
             // kilocode_change start - pass turn context for slow-snapshot UI/policy handling
             const completedSnapshot = yield* snapshot.track({
               sessionID: ctx.sessionID,
@@ -1005,7 +1028,10 @@ export const layer = Layer.effect(
               },
               { text: ctx.currentText.text },
             )).text
-            if (ctx.currentText.text.trim()) ctx.step.text = true // kilocode_change
+            if (ctx.currentText.text.trim()) attempt.text = true // kilocode_change
+            if (ctx.currentText.text.trim()) {
+              ctx.step.text = true
+            } // kilocode_change
             if (!ctx.assistantMessage.summary) {
               // TODO(v2): Temporary dual-write while migrating session messages to v2 events.
               if (mirrorAssistant) {
@@ -1124,6 +1150,7 @@ export const layer = Layer.effect(
         slog.error("process", { error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined })
         const error = parse(e)
         // kilocode_change start
+        if (e instanceof KiloSessionProcessor.IncompleteResponseError) ctx.assistantMessage.finish = "unknown"
         ctx.compactionError = MessageV2.ContextOverflowError.isInstance(error) ? error : ctx.compactionError
         // kilocode_change end
         yield* flushV2Fragments()
@@ -1182,76 +1209,142 @@ export const layer = Layer.effect(
         ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true
 
         return yield* Effect.gen(function* () {
-          yield* Effect.gen(function* () {
+          // kilocode_change start - publish retry state consistently for provider and empty-response retries
+          const retries = { provider: 0 }
+          const setRetry = (info: {
+            attempt: number
+            message: string
+            action?: SessionRetry.Retryable["action"]
+            next: number
+          }) => {
+            const event = mirrorAssistant
+              ? events.publish(SessionEvent.Retried, {
+                  sessionID: ctx.sessionID,
+                  attempt: info.attempt,
+                  error: {
+                    message: info.message,
+                    isRetryable: true,
+                  },
+                  timestamp: DateTime.makeUnsafe(Date.now()),
+                })
+              : Effect.void
+            return flushV2Fragments().pipe(
+              Effect.andThen(event),
+              Effect.andThen(
+                status.set(ctx.sessionID, {
+                  type: "retry",
+                  attempt: info.attempt,
+                  message: info.message,
+                  action: info.action,
+                  next: info.next,
+                }),
+              ),
+            )
+          }
+
+          const request = () =>
+            Effect.gen(function* () {
+              ctx.currentText = undefined
+              ctx.currentTextID = undefined
+              ctx.reasoningMap = {}
+              yield* status.set(ctx.sessionID, { type: "busy" })
+              ctx.step = { reasoning: false, text: false, tool: false }
+              const stream = llm.stream({
+                ...streamInput,
+                preflight: !ctx.assistantMessage.summary,
+              })
+
+              yield* stream.pipe(
+                Stream.tap((event) => handleEvent(event)),
+                Stream.takeUntil(() => ctx.needsCompaction),
+                Stream.runDrain,
+              )
+            }).pipe(
+              Effect.onInterrupt(() =>
+                Effect.gen(function* () {
+                  aborted = true
+                  ac.abort() // kilocode_change — also abort offline handler
+                  if (!ctx.assistantMessage.error) {
+                    yield* halt(new DOMException("Aborted", "AbortError"))
+                  }
+                }),
+              ),
+              Effect.catchCauseIf(
+                (cause) => !Cause.hasInterruptsOnly(cause),
+                (cause) => Effect.fail(Cause.squash(cause)),
+              ),
+              Effect.retry(
+                SessionRetry.policy({
+                  provider: input.model.providerID,
+                  parse: retryParse,
+                  ...KiloSessionProcessor.retryOpts({
+                    sessionID: ctx.sessionID,
+                    abort: ac.signal,
+                    set: status.set,
+                    used: retries.provider,
+                  }),
+                  set: (info) => {
+                    if (info.attempt > 0) retries.provider += 1
+                    return setRetry(info)
+                  },
+                }),
+              ),
+            )
+
+          const discard = Effect.fn("SessionProcessor.discardIncomplete")(function* (baseline: Set<string>) {
+            yield* Effect.forEach(
+              Object.values(ctx.toolcalls),
+              (call) => Deferred.succeed(call.done, undefined).pipe(Effect.ignore),
+              { concurrency: "unbounded" },
+            )
+            const parts = yield* MessageV2.parts(ctx.assistantMessage.id).pipe(
+              Effect.provideService(Database.Service, database),
+            )
+            yield* Effect.forEach(
+              parts.filter((part) => !baseline.has(part.id)),
+              (part) =>
+                session.removePart({ sessionID: ctx.sessionID, messageID: ctx.assistantMessage.id, partID: part.id }),
+              { concurrency: 1 },
+            )
             ctx.currentText = undefined
             ctx.currentTextID = undefined
             ctx.reasoningMap = {}
-            yield* status.set(ctx.sessionID, { type: "busy" })
-            // kilocode_change start
-            ctx.step = { reasoning: false, text: false, tool: false }
-            const stream = llm.stream({
-              ...streamInput,
-              preflight: !ctx.assistantMessage.summary,
-            })
-            // kilocode_change end
+            ctx.toolcalls = {}
+            ctx.toolmeta = {}
+            ctx.assistantMessage.finish = undefined
+          })
 
-            yield* stream.pipe(
-              Stream.tap((event) => handleEvent(event)),
-              Stream.takeUntil(() => ctx.needsCompaction),
-              Stream.runDrain,
-            )
-          }).pipe(
-            Effect.onInterrupt(() =>
-              Effect.gen(function* () {
-                aborted = true
-                ac.abort() // kilocode_change — also abort offline handler
-                if (!ctx.assistantMessage.error) {
-                  yield* halt(new DOMException("Aborted", "AbortError"))
-                }
+          const recover = () => {
+            const baseline = new Set<string>()
+            return KiloSessionProcessor.recover({
+              run: Effect.fn("SessionProcessor.incompleteAttempt")(function* () {
+                baseline.clear()
+                for (const part of yield* MessageV2.parts(ctx.assistantMessage.id).pipe(
+                  Effect.provideService(Database.Service, database),
+                ))
+                  baseline.add(part.id)
+                attempt = KiloSessionProcessor.attempt()
+                yield* request()
               }),
-            ),
-            Effect.catchCauseIf(
-              (cause) => !Cause.hasInterruptsOnly(cause),
-              (cause) => Effect.fail(Cause.squash(cause)),
-            ),
-            Effect.retry(
-              SessionRetry.policy({
-                provider: input.model.providerID,
-                parse,
-                // kilocode_change start
-                ...KiloSessionProcessor.retryOpts({ sessionID: ctx.sessionID, abort: ac.signal, set: status.set }),
-                // kilocode_change end
-                set: (info) => {
-                  // TODO(v2): Temporary dual-write while migrating session messages to v2 events.
-                  const event = mirrorAssistant
-                    ? events.publish(SessionEvent.Retried, {
-                        sessionID: ctx.sessionID,
-                        attempt: info.attempt,
-                        error: {
-                          message: info.message,
-                          isRetryable: true,
-                        },
-                        timestamp: DateTime.makeUnsafe(Date.now()),
-                      })
-                    : Effect.void
-                  return flushV2Fragments().pipe(
-                    Effect.andThen(event),
-                    Effect.andThen(
-                      status.set(ctx.sessionID, {
-                        type: "retry",
-                        attempt: info.attempt,
-                        message: info.message,
-                        action: info.action,
-                        next: info.next,
-                      }),
-                    ),
-                  )
-                },
-              }),
-            ),
+              replayable: () =>
+                !mirrorAssistant &&
+                KiloSessionProcessor.replayable({
+                  finish: attempt.finish,
+                  text: attempt.text,
+                  reasoning: attempt.reasoning,
+                  tool: attempt.tool,
+                  usage: attempt.usage,
+                }),
+              discard: () => discard(baseline),
+              set: setRetry,
+            })
+          }
+
+          yield* recover().pipe(
             Effect.catch(halt),
             Effect.ensuring(cleanup()),
           )
+          // kilocode_change end
 
           if (ctx.needsCompaction) return "compact"
           if (ctx.blocked || ctx.assistantMessage.error) return "stop"

packages/opencode/test/kilocode/session-processor-incomplete-response-retry.test.ts

@@ -0,0 +1,578 @@
+import { NodeFileSystem } from "@effect/platform-node"
+import { describe, expect, spyOn } from "bun:test"
+import { APICallError } from "ai"
+import { Context, Effect, Layer } from "effect"
+import * as Stream from "effect/Stream"
+import { LLMEvent, Usage, type LLMEvent as Event } from "@opencode-ai/llm"
+import { Database } from "@opencode-ai/core/database/database"
+import path from "path"
+import { Agent as AgentSvc } from "../../src/agent/agent"
+import { Bus } from "../../src/bus"
+import { Config } from "../../src/config/config"
+import { RuntimeFlags } from "../../src/effect/runtime-flags"
+import { EventV2Bridge } from "../../src/event-v2-bridge"
+import { Image } from "../../src/image/image"
+import { KiloSessionProcessor } from "../../src/kilocode/session/processor"
+import { Permission } from "../../src/permission"
+import { Plugin } from "../../src/plugin"
+import type { Provider } from "../../src/provider/provider"
+import { ProviderV2 } from "@opencode-ai/core/provider"
+import { ModelV2 } from "@opencode-ai/core/model"
+import { Reference } from "../../src/reference/reference"
+import { Session } from "../../src/session/session"
+import { LLM } from "../../src/session/llm"
+import { MessageV2 } from "../../src/session/message-v2"
+import { SessionProcessor } from "../../src/session/processor"
+import { SessionRetry } from "../../src/session/retry"
+import { MessageID } from "../../src/session/schema"
+import { SessionStatus } from "../../src/session/status"
+import { SessionSummary } from "../../src/session/summary"
+import { Snapshot } from "../../src/snapshot"
+import { SyncEvent } from "../../src/sync"
+import * as Log from "@opencode-ai/core/util/log"
+import * as CrossSpawnSpawner from "@opencode-ai/core/cross-spawn-spawner"
+import { provideTmpdirProject } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+Log.init({ print: false })
+
+const ref = {
+  providerID: ProviderV2.ID.make("test"),
+  modelID: ModelV2.ID.make("test-model"),
+}
+
+type Script = Stream.Stream<Event, unknown>
+
+class TestLLM extends Context.Service<
+  TestLLM,
+  {
+    readonly push: (stream: Script) => Effect.Effect<void>
+    readonly reply: (...events: Event[]) => Effect.Effect<void>
+    readonly calls: Effect.Effect<number>
+  }
+>()("@test/IncompleteResponseRetryLLM") {}
+
+function model(): Provider.Model {
+  return {
+    id: ref.modelID,
+    providerID: ref.providerID,
+    name: "Test",
+    limit: { context: 128000, output: 4096 },
+    cost: { input: 0, output: 0, cache: { read: 0, write: 0 } },
+    capabilities: {
+      toolcall: true,
+      attachment: false,
+      reasoning: true,
+      temperature: true,
+      input: { text: true, image: false, audio: false, video: false },
+      output: { text: true, image: false, audio: false, video: false },
+    },
+    api: { npm: "@ai-sdk/openai" },
+    options: {},
+  } as Provider.Model
+}
+
+function empty() {
+  const usage = new Usage({})
+  return [
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.reasoningStart({ id: "reasoning" }),
+    LLMEvent.reasoningEnd({ id: "reasoning" }),
+    LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+    LLMEvent.finish({ reason: "unknown", usage }),
+  ]
+}
+
+function success() {
+  const usage = new Usage({ inputTokens: 10, outputTokens: 2, totalTokens: 12 })
+  return [
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.textStart({ id: "text" }),
+    LLMEvent.textDelta({ id: "text", text: "Recovered" }),
+    LLMEvent.textEnd({ id: "text" }),
+    LLMEvent.stepFinish({ index: 0, reason: "stop", usage }),
+    LLMEvent.finish({ reason: "stop", usage }),
+  ]
+}
+
+function retryable429() {
+  return new APICallError({
+    message: "429 status code (no body)",
+    url: "https://example.test/v1/chat/completions",
+    requestBodyValues: {},
+    statusCode: 429,
+    responseHeaders: { "content-type": "application/json" },
+    isRetryable: true,
+  })
+}
+
+const llm = Layer.unwrap(
+  Effect.gen(function* () {
+    const queue: Script[] = []
+    let calls = 0
+    const push = (stream: Script) => {
+      queue.push(stream)
+      return Effect.void
+    }
+    return Layer.mergeAll(
+      Layer.succeed(
+        LLM.Service,
+        LLM.Service.of({
+          stream: () => {
+            calls += 1
+            return queue.shift() ?? Stream.fail(new Error("unexpected extra llm call"))
+          },
+        }),
+      ),
+      Layer.succeed(
+        TestLLM,
+        TestLLM.of({
+          push,
+          reply: (...events) => push(Stream.make(...events)),
+          calls: Effect.sync(() => calls),
+        }),
+      ),
+    )
+  }),
+)
+
+const reference = Layer.mock(Reference.Service)({
+  init: () => Effect.void,
+  list: () => Effect.succeed([]),
+  get: () => Effect.succeed(undefined),
+  ensure: () => Effect.void,
+  contains: () => Effect.succeed(false),
+})
+const status = Layer.mergeAll(SessionStatus.defaultLayer, Bus.layer)
+const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer)
+const env = (event = false) =>
+  SessionProcessor.layer.pipe(
+    Layer.provideMerge(
+      Layer.mergeAll(
+        Session.defaultLayer,
+        Snapshot.defaultLayer,
+        AgentSvc.defaultLayer,
+        Permission.defaultLayer,
+        Plugin.defaultLayer,
+        Config.defaultLayer,
+        RuntimeFlags.layer({ experimentalEventSystem: event }),
+        reference,
+        SessionSummary.defaultLayer,
+        Image.defaultLayer,
+        SyncEvent.defaultLayer,
+        EventV2Bridge.defaultLayer,
+        Database.defaultLayer,
+        status,
+        llm,
+      ).pipe(Layer.provideMerge(infra)),
+    ),
+    Layer.provide(reference),
+  )
+
+const it = testEffect(env())
+const eventIt = testEffect(env(true))
+
+const setup = Effect.fn("SessionProcessorIncompleteRetryTest.setup")(function* (dir: string) {
+  const test = yield* TestLLM
+  const processors = yield* SessionProcessor.Service
+  const session = yield* Session.Service
+  const chat = yield* session.create({})
+  const parent = yield* session.updateMessage({
+    id: MessageID.ascending(),
+    role: "user",
+    sessionID: chat.id,
+    agent: "code",
+    model: ref,
+    time: { created: Date.now() },
+  })
+  const msg: MessageV2.Assistant = {
+    id: MessageID.ascending(),
+    role: "assistant",
+    sessionID: chat.id,
+    parentID: parent.id,
+    mode: "code",
+    agent: "code",
+    path: { cwd: path.resolve(dir), root: path.resolve(dir) },
+    cost: 0,
+    tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
+    modelID: ref.modelID,
+    providerID: ref.providerID,
+    time: { created: Date.now() },
+  }
+  yield* session.updateMessage(msg)
+  const mdl = model()
+  const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl })
+  const input: LLM.StreamInput = {
+    user: parent as MessageV2.User,
+    sessionID: chat.id,
+    model: mdl,
+    agent: { name: "code", mode: "primary", permission: [], options: {} } as any,
+    system: [],
+    messages: [],
+    tools: {},
+  }
+  return { test, session, msg, handle, input }
+})
+
+describe("session processor incomplete response retry", () => {
+  it.effect("retries an empty unknown response and removes the failed attempt", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.reply(...success())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          } finally {
+            delay.mockRestore()
+          }
+
+          expect(yield* ctx.test.calls).toBe(2)
+          expect(ctx.handle.message.finish).toBe("stop")
+          const parts = yield* MessageV2.parts(ctx.msg.id)
+          expect(parts.map((part) => part.type)).toEqual(["step-start", "text", "step-finish"])
+          expect(parts.some((part) => part.type === "reasoning")).toBe(false)
+          expect(parts.find((part) => part.type === "text")?.text).toBe("Recovered")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("stops after two empty response retries", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.push(Stream.fail(new Error("unexpected extra llm call")))
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("stop")
+          } finally {
+            delay.mockRestore()
+          }
+
+          expect(yield* ctx.test.calls).toBe(3)
+          expect(ctx.handle.message.finish).toBe("unknown")
+          const error = ctx.handle.message.error
+          expect(MessageV2.APIError.isInstance(error)).toBe(true)
+          if (!MessageV2.APIError.isInstance(error)) throw new Error("expected API error")
+          expect(error.data.message).toBe(KiloSessionProcessor.INCOMPLETE_RESPONSE_MESSAGE)
+          expect(yield* MessageV2.parts(ctx.msg.id)).toEqual([])
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("retries when the stream drains without a finish event", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(LLMEvent.stepStart({ index: 0 }))
+          yield* ctx.test.reply(...success())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          } finally {
+            delay.mockRestore()
+          }
+
+          expect(yield* ctx.test.calls).toBe(2)
+          expect((yield* MessageV2.parts(ctx.msg.id)).map((part) => part.type)).toEqual([
+            "step-start",
+            "text",
+            "step-finish",
+          ])
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry non-empty reasoning", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.reasoningStart({ id: "reasoning-1" }),
+            LLMEvent.reasoningDelta({ id: "reasoning-1", text: "Investigating the problem" }),
+            LLMEvent.reasoningEnd({ id: "reasoning-1" }),
+            LLMEvent.reasoningStart({ id: "reasoning-2" }),
+            LLMEvent.reasoningDelta({ id: "reasoning-2", text: "Preparing the final answer" }),
+            LLMEvent.reasoningEnd({ id: "reasoning-2" }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "unknown", usage }),
+          )
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.finish).toBe("unknown")
+          const parts = yield* MessageV2.parts(ctx.msg.id)
+          expect(parts.filter((part) => part.type === "reasoning").map((part) => part.text)).toEqual([
+            "Investigating the problem",
+            "Preparing the final answer",
+          ])
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("removes incomplete tool framing before retrying", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.toolInputStart({ id: "call", name: "write" }),
+            LLMEvent.toolInputDelta({ id: "call", name: "write", text: "{\"path\":" }),
+          )
+          yield* ctx.test.reply(...success())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          } finally {
+            delay.mockRestore()
+          }
+
+          expect(yield* ctx.test.calls).toBe(2)
+          const parts = yield* MessageV2.parts(ctx.msg.id)
+          expect(parts.some((part) => part.type === "tool")).toBe(false)
+          expect(parts.find((part) => part.type === "text")?.text).toBe("Recovered")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry partial visible output", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.textStart({ id: "partial" }),
+            LLMEvent.textDelta({ id: "partial", text: "Partial" }),
+            LLMEvent.textEnd({ id: "partial" }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "unknown", usage }),
+          )
+
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.finish).toBe("unknown")
+          expect((yield* MessageV2.parts(ctx.msg.id)).find((part) => part.type === "text")?.text).toBe("Partial")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry a complete tool call", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.toolCall({ id: "call", name: "web_search", input: { query: "Kilo" }, providerExecuted: true }),
+            LLMEvent.toolResult({
+              id: "call",
+              name: "web_search",
+              result: { type: "json", value: { output: "result" } },
+              providerExecuted: true,
+            }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "unknown", usage }),
+          )
+
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          expect(yield* ctx.test.calls).toBe(1)
+          expect((yield* MessageV2.parts(ctx.msg.id)).some((part) => part.type === "tool")).toBe(true)
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry an unmatched tool error", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.toolError({ id: "missing", name: "web_search", message: "provider tool failed" }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "unknown", usage }),
+          )
+
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.finish).toBe("unknown")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry an unknown finish with non-zero usage", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({ inputTokens: 10, outputTokens: 1, totalTokens: 11 })
+          const empty = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "unknown", usage: empty }),
+          )
+
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.finish).toBe("unknown")
+          expect(ctx.handle.message.tokens.input).toBe(10)
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("uses step finish instead of a conflicting final finish", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          const usage = new Usage({})
+          yield* ctx.test.reply(
+            LLMEvent.stepStart({ index: 0 }),
+            LLMEvent.reasoningStart({ id: "reasoning" }),
+            LLMEvent.reasoningEnd({ id: "reasoning" }),
+            LLMEvent.stepFinish({ index: 0, reason: "unknown", usage }),
+            LLMEvent.finish({ reason: "stop", usage }),
+          )
+          yield* ctx.test.reply(...success())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          } finally {
+            delay.mockRestore()
+          }
+
+          expect(yield* ctx.test.calls).toBe(2)
+          expect(ctx.handle.message.finish).toBe("stop")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("keeps provider retries independent after an empty response", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          process.env.KILO_SESSION_RETRY_LIMIT = "2"
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.push(Stream.fail(retryable429()))
+          yield* ctx.test.push(Stream.fail(retryable429()))
+          yield* ctx.test.reply(...success())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          } finally {
+            delay.mockRestore()
+            delete process.env.KILO_SESSION_RETRY_LIMIT
+          }
+
+          expect(yield* ctx.test.calls).toBe(4)
+          expect(ctx.handle.message.finish).toBe("stop")
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("does not retry a provider error after final output", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          process.env.KILO_SESSION_RETRY_LIMIT = "1"
+          const ctx = yield* setup(dir)
+          yield* ctx.test.push(
+            Stream.make(
+              LLMEvent.stepStart({ index: 0 }),
+              LLMEvent.textStart({ id: "partial" }),
+              LLMEvent.textDelta({ id: "partial", text: "Partial" }),
+            ).pipe(Stream.concat(Stream.fail(retryable429()))),
+          )
+          yield* ctx.test.reply(...empty())
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("stop")
+          } finally {
+            delay.mockRestore()
+            delete process.env.KILO_SESSION_RETRY_LIMIT
+          }
+
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.error).toBeDefined()
+          expect((yield* MessageV2.parts(ctx.msg.id)).some((part) => part.type === "text")).toBe(true)
+        }),
+      { git: true },
+    ),
+  )
+
+  it.effect("keeps the provider retry budget cumulative across incomplete retries", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          process.env.KILO_SESSION_RETRY_LIMIT = "2"
+          const ctx = yield* setup(dir)
+          yield* ctx.test.push(Stream.fail(retryable429()))
+          yield* ctx.test.reply(...empty())
+          yield* ctx.test.push(Stream.fail(retryable429()))
+          yield* ctx.test.push(Stream.fail(retryable429()))
+          const delay = spyOn(SessionRetry, "delay").mockReturnValue(0)
+
+          try {
+            expect(yield* ctx.handle.process(ctx.input)).toBe("stop")
+          } finally {
+            delay.mockRestore()
+            delete process.env.KILO_SESSION_RETRY_LIMIT
+          }
+
+          expect(yield* ctx.test.calls).toBe(4)
+          expect(MessageV2.APIError.isInstance(ctx.handle.message.error)).toBe(true)
+        }),
+      { git: true },
+    ),
+  )
+
+  eventIt.effect("does not retry when Event V2 mirroring is enabled", () =>
+    provideTmpdirProject(
+      (dir) =>
+        Effect.gen(function* () {
+          const ctx = yield* setup(dir)
+          yield* ctx.test.reply(...empty())
+
+          expect(yield* ctx.handle.process(ctx.input)).toBe("continue")
+          expect(yield* ctx.test.calls).toBe(1)
+          expect(ctx.handle.message.finish).toBe("unknown")
+        }),
+      { git: true },
+    ),
+  )
+})