fix(session): recover from stalled model requests

docs/docs/configure/providers.md

@@ -503,3 +503,22 @@ The `small_model` is used for lightweight tasks like summarization and context c
 | `options.baseURL` | `string` | Custom endpoint URL for Bedrock gateway/proxy (Amazon Bedrock only) |
 | `project` | `string` | GCP project ID (Google Vertex AI only) |
 | `location` | `string` | GCP region (Google Vertex AI only, default: `us-central1`) |
+| `options.headerTimeout` | `number \| false` | Milliseconds to wait for the response headers (the first byte) before the request is aborted and retried (default `300000`; the OpenAI provider keeps its own `10000`). `false` turns the check off, and a server that then never answers hangs the session |
+| `options.chunkTimeout` | `number` | Milliseconds of silence on a streaming response before the request is aborted and retried (default `300000`). Applies to SSE and Amazon Bedrock event streams |
+
+### Stalled requests
+
+If the provider accepts a request and then stops answering, Altimate Code aborts the connection when
+`headerTimeout` or `chunkTimeout` elapses and retries the request with the usual backoff, up to five
+times. Whatever the stalled attempt had already streamed (text, reasoning, a tool call still being
+written) is discarded so the retry does not duplicate it. A tool call that had already been dispatched
+is never re-run: the session stops with "The model stopped responding" and names the tool instead of
+retrying. When retries run out the session ends with the same message.
+
+```json
+{
+  "provider": {
+    "amazon-bedrock": { "options": { "headerTimeout": 120000, "chunkTimeout": 120000 } }
+  }
+}
+```

packages/opencode/src/provider/error.ts

@@ -19,6 +19,11 @@ export namespace ProviderError {
     }
   }
 
+  // Message of the idle-stream watchdog, shared so classification can recognise it.
+  export const SSE_IDLE_MESSAGE = "SSE read timed out"
+  // Leading words of the user-facing message for a stalled request.
+  export const MODEL_STOPPED_RESPONDING = "The model stopped responding"
+
   export class ResponseStreamError extends Error {
     public override readonly name = "ProviderResponseStreamError"
 

packages/opencode/src/provider/provider.ts

@@ -82,6 +82,14 @@ import { isValidDatabricksHost } from "../altimate/plugin/databricks"
 // altimate_change start — raise SSE chunk-timeout watchdog 2min→5min (#844) for slow warehouse/LLM streams
 const DEFAULT_CHUNK_TIMEOUT = 300_000
 // altimate_change end
+// altimate_change start — a request the server never answers must not hang forever: default first-byte
+// (response headers) timeout for providers that set none of their own. 300s matches upstream OpenCode
+// (dev #46903, which raised shorter defaults after false positives) and the chunk watchdog below. In 6,586
+// recorded agent steps the slowest healthy generation took 51s, so no healthy request there is near it,
+// while the 10 observed hangs were never answered at all. Override per provider with
+// `provider.<id>.options.headerTimeout` (ms, or false to disable).
+const DEFAULT_HEADER_TIMEOUT = 300_000
+// altimate_change end
 // altimate_change start — OpenAI-compatible HTTP header timeout support
 const OPENAI_HEADER_TIMEOUT_DEFAULT = 10_000
 const HEADER_TIMEOUT = Symbol.for("opencode.provider.header-timeout")
@@ -141,15 +149,19 @@ export namespace Provider {
   function wrapSSE(res: Response, ms: number, ctl: AbortController) {
     if (typeof ms !== "number" || ms <= 0) return res
     if (!res.body) return res
-    if (!res.headers.get("content-type")?.includes("text/event-stream")) return res
+    // altimate_change start — Bedrock Converse streams binary event-stream frames, not SSE; watch those too
+    const contentType = (res.headers.get("content-type") ?? "").toLowerCase()
+    if (!contentType.includes("text/event-stream") && !contentType.includes("application/vnd.amazon.eventstream"))
+      return res
+    // altimate_change end
 
     const reader = res.body.getReader()
     const body = new ReadableStream<Uint8Array>({
       async pull(ctrl) {
         const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
           const id = setTimeout(() => {
             // altimate_change start — surface stalled SSE reads as ProviderError.ResponseStreamError
-            const err = new ProviderError.ResponseStreamError("SSE read timed out")
+            const err = new ProviderError.ResponseStreamError(ProviderError.SSE_IDLE_MESSAGE)
             // altimate_change end
             ctl.abort(err)
             void reader.cancel(err)
@@ -1964,7 +1976,9 @@ export namespace Provider {
 
       const customFetch = options["fetch"]
       const chunkTimeout = options["chunkTimeout"] || DEFAULT_CHUNK_TIMEOUT
-      const headerTimeout = options["headerTimeout"]
+      // altimate_change start — first-byte timeout default (see DEFAULT_HEADER_TIMEOUT)
+      const headerTimeout = options["headerTimeout"] ?? DEFAULT_HEADER_TIMEOUT
+      // altimate_change end
       delete options["chunkTimeout"]
       delete options["headerTimeout"]
 
@@ -2014,12 +2028,38 @@ export namespace Provider {
           }
         }
 
-        const res = await (async () =>
+        // altimate_change start — the deadline must also bound a custom fetch's own pre-request work (for example
+        // Vertex's credential acquisition), which never observes the signal: race it against the deadline.
+        const pending = (async () =>
           fetchFn(input, {
             ...opts,
             // @ts-ignore see here: https://github.com/oven-sh/bun/issues/16682
             timeout: false,
-          }))().finally(() => headerTimeoutCtl?.clear())
+          }))()
+        let res: Response
+        if (opts.signal) {
+          // settle on the combined abort (first-byte deadline, caller cancel, timeouts), whichever comes first
+          const guards = [opts.signal]
+          const cleanups: Array<() => void> = []
+          const deadline = new Promise<never>((_, reject) => {
+            for (const guard of guards) {
+              const onAbort = () => reject(guard.reason)
+              if (guard.aborted) return onAbort()
+              guard.addEventListener("abort", onAbort, { once: true })
+              cleanups.push(() => guard.removeEventListener("abort", onAbort))
+            }
+          })
+          pending.catch(() => {}) // the loser of the race must not surface as an unhandled rejection
+          try {
+            res = await Promise.race([pending, deadline])
+          } finally {
+            for (const cleanup of cleanups) cleanup()
+            headerTimeoutCtl?.clear()
+          }
+        } else {
+          res = await pending.finally(() => headerTimeoutCtl?.clear())
+        }
+        // altimate_change end
 
         if (!chunkAbortCtl) return res
         return wrapSSE(res, chunkTimeout, chunkAbortCtl)

packages/opencode/src/session/message-v2.ts

@@ -1171,6 +1171,30 @@ export namespace MessageV2 {
           },
           { cause: e },
         ).toObject()
+      // altimate_change start — a request the provider stopped answering (no response headers, or a silent
+      // stream) is a transient stall: classify it as a retryable APIError that names the problem instead of
+      // surfacing it as an opaque UnknownError that ends the session. Mirrors upstream OpenCode
+      // (dev: message-v2.ts HeaderTimeoutError / ResponseStreamError cases), with clearer wording and
+      // limited to the idle watchdog: other ResponseStreamErrors (websocket transport) keep their old handling.
+      case e instanceof ProviderError.HeaderTimeoutError:
+        return new MessageV2.APIError(
+          {
+            message: `${ProviderError.MODEL_STOPPED_RESPONDING}: no response headers within ${Math.round(e.ms / 1000)}s`,
+            isRetryable: true,
+            metadata: { code: e.name, timeoutMs: String(e.ms) },
+          },
+          { cause: e },
+        ).toObject()
+      case e instanceof ProviderError.ResponseStreamError && e.message === ProviderError.SSE_IDLE_MESSAGE:
+        return new MessageV2.APIError(
+          {
+            message: `${ProviderError.MODEL_STOPPED_RESPONDING}: the response stream went silent (${e.message})`,
+            isRetryable: true,
+            metadata: { code: e.name },
+          },
+          { cause: e },
+        ).toObject()
+      // altimate_change end
       case APICallError.isInstance(e):
         const parsed = ProviderError.parseAPICallError({
           providerID: ctx.providerID,

packages/opencode/src/session/processor.ts

@@ -9,6 +9,10 @@ import { Snapshot } from "@/snapshot"
 import { SessionSummary } from "./summary"
 import { Bus } from "@/bus"
 import { SessionRetry } from "./retry"
+// altimate_change start — discard a failed attempt's partial output before re-requesting
+import { StallRecovery } from "./stall-recovery"
+import { ProviderError } from "@/provider/error"
+// altimate_change end
 import { SessionStatus } from "./status"
 import { Plugin } from "@/plugin"
 import type { Provider } from "@/provider/provider"
@@ -233,6 +237,22 @@ export namespace SessionProcessor {
       executionID(execution: ToolExecution) {
         return allocated.get(keyOf(execution.raw))?.[execution.occurrence]
       },
+      // Forget tool inputs that started streaming but were never called (a discarded attempt) as if they never
+      // happened: drop their allocations so the per-id slot ordinals that settled()/executionID() index by stay
+      // aligned, and free their names. Ids allocated by earlier calls stay reserved, so none is ever reused.
+      discardUnstarted() {
+        for (const [key, ids] of started) {
+          const list = allocated.get(key) ?? []
+          for (const id of ids) {
+            const index = list.lastIndexOf(id)
+            if (index >= 0) list.splice(index, 1)
+            used.delete(id)
+            occurrences.set(key, Math.max(0, (occurrences.get(key) ?? 1) - 1))
+          }
+          if (list.length === 0) allocated.delete(key)
+        }
+        started.clear()
+      },
     })
   }
   // altimate_change end
@@ -283,6 +303,7 @@ export namespace SessionProcessor {
     // session thinking it's stuck. Track whether the session has ever produced a
     // tool call; if plan agent finishes its first step with stop-no-tools, warn.
     let sessionToolCallsMade = 0
+    let toolExecutionsStarted = 0
     let planNoToolWarningEmitted = false
     // altimate_change end
 
@@ -300,6 +321,10 @@ export namespace SessionProcessor {
         return toolcalls.get(matched ?? coerceToolCallID.peek(toolCallID))
       },
       beginToolExecution(toolCallID: string) {
+        // altimate_change start — the SDK runs a tool as soon as its input is complete, possibly before its part
+        // leaves `pending` (or is persisted at all): this counter is the source of truth for "a tool has started"
+        toolExecutionsStarted++
+        // altimate_change end
         return coerceToolCallID.beginExecution(toolCallID)
       },
       finishToolExecution(execution: ToolExecution) {
@@ -408,7 +433,14 @@ export namespace SessionProcessor {
         }
         // altimate_change end
         while (true) {
+          // altimate_change start — remember which parts predate this attempt so a failed one can be unwound
+          let partsBeforeAttempt = new Set<string>()
+          const executionsBeforeAttempt = toolExecutionsStarted
+          // altimate_change end
           try {
+            // altimate_change start — see above
+            partsBeforeAttempt = StallRecovery.partIDs(input.assistantMessage.id)
+            // altimate_change end
             let currentText: MessageV2.TextPart | undefined
             let reasoningMap: Record<string, MessageV2.ReasoningPart> = {}
             if (snapshot === undefined) {
@@ -1195,8 +1227,29 @@ export namespace SessionProcessor {
               }
             } else {
               const retry = SessionRetry.retryable(error)
-              // altimate_change start — cap retries to avoid infinite loops, log on exhaustion
-              if (retry !== undefined && attempt < SessionRetry.RETRY_MAX_ATTEMPTS) {
+              // altimate_change start — cap retries to avoid infinite loops, log on exhaustion.
+              // A user cancel is never retried; and a retry only goes ahead after the failed attempt's
+              // partial output is discarded, which is refused once a tool call was dispatched (see
+              // StallRecovery) so a retry can neither duplicate streamed output nor re-run a tool.
+              const discard =
+                retry !== undefined && attempt < SessionRetry.RETRY_MAX_ATTEMPTS && !input.abort.aborted
+                  ? await StallRecovery.discardAttempt({
+                      sessionID: input.sessionID,
+                      messageID: input.assistantMessage.id,
+                      before: partsBeforeAttempt,
+                      toolExecutionStarted: toolExecutionsStarted > executionsBeforeAttempt,
+                    }).catch((err) => {
+                      // a failed cleanup must not escape the catch block: fall through to the terminal path
+                      log.warn("could not discard partial output; not retrying", { error: err })
+                      return undefined
+                    })
+                  : undefined
+              if (discard && !discard.ok) {
+                log.warn("not retrying: the attempt already acted", { parts: discard.dispatched })
+              }
+              if (retry !== undefined && discard?.ok) {
+                for (const [id, part] of toolcalls) if (discard.removed.has(part.id)) toolcalls.delete(id)
+                coerceToolCallID.discardUnstarted()
                 // altimate_change end
                 attempt++
                 const delay = SessionRetry.delay(attempt, error.name === "APIError" ? error : undefined)
@@ -1212,7 +1265,8 @@ export namespace SessionProcessor {
                 continue
               }
               // altimate_change start — log when retries exhausted for debugging
-              if (retry !== undefined) {
+              const exhausted = retry !== undefined && attempt >= SessionRetry.RETRY_MAX_ATTEMPTS
+              if (exhausted) {
                 log.warn("max retry attempts reached, giving up", {
                   attempt,
                   message: retry,
@@ -1221,6 +1275,15 @@ export namespace SessionProcessor {
                 })
               }
               // altimate_change end
+              // altimate_change start — say why a retryable error was not retried or retried out
+              if (MessageV2.APIError.isInstance(error)) {
+                if (discard && !discard.ok) {
+                  error.data.message += ` (not retried: the attempt had already acted: ${discard.dispatched.join(", ")})`
+                } else if (exhausted && error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING)) {
+                  error.data.message += ` (gave up after ${attempt} retries)`
+                }
+              }
+              // altimate_change end
               input.assistantMessage.error = error
               // altimate_change start — order terminal error before idle publication
               // Publish the error before idle. The run harness drains this

packages/opencode/src/session/retry.ts

@@ -13,6 +13,10 @@ export namespace SessionRetry {
 
   export async function sleep(ms: number, signal: AbortSignal): Promise<void> {
     return new Promise((resolve, reject) => {
+      // altimate_change start — upstream_fix: an already-aborted signal never fires its listener, so a Stop that
+      // landed just before the backoff began would otherwise wait out the whole delay
+      if (signal.aborted) return reject(new DOMException("Aborted", "AbortError"))
+      // altimate_change end
       const abortHandler = () => {
         clearTimeout(timeout)
         reject(new DOMException("Aborted", "AbortError"))

packages/opencode/src/session/stall-recovery.ts

@@ -0,0 +1,78 @@
+import { Log } from "@/util/log"
+import { MessageV2 } from "./message-v2"
+import { Bus } from "@/bus"
+import { Database, and, eq, inArray } from "@/storage/db"
+import { PartTable } from "@opencode-ai/core/session/sql"
+import type { MessageID, SessionID } from "./schema"
+
+/**
+ * Cleanup that makes it safe to re-issue a model request inside the same assistant message after the
+ * previous attempt failed part-way (stalled, reset, or errored mid-stream).
+ *
+ * Rule: a retry may only replace what the failed attempt streamed, never what it already acted on.
+ *  - text, reasoning and step-start parts written by the failed attempt are removed, and so are tool parts
+ *    still `pending` (input was streaming, the tool never started). The retried request regenerates them.
+ *  - if the failed attempt left any tool part that STARTED (running, completed or errored), the tool call was
+ *    dispatched and may have had side effects. Re-requesting would make the model emit it again, so the attempt
+ *    is NOT retried and nothing is removed: the error surfaces and the dispatched call keeps its record.
+ *  - the same holds if the attempt already finished its step (a step-finish part exists): its cost and tokens
+ *    are on the message, so a retry would count them twice and answer a second time.
+ */
+const log = Log.create({ service: "session.stall-recovery" })
+
+// A `tool` part reaching this set is necessarily still `pending` (started ones are rejected first).
+const DISCARDED_WHEN_RETRIED: ReadonlySet<MessageV2.Part["type"]> = new Set(["text", "reasoning", "step-start", "tool"])
+
+export function partIDs(messageID: MessageID): Set<string> {
+  return new Set(MessageV2.parts(messageID).map((part) => part.id))
+}
+
+export type Discard = { ok: true; removed: Set<string> } | { ok: false; dispatched: string[] }
+
+export async function discardAttempt(input: {
+  sessionID: SessionID
+  messageID: MessageID
+  before: ReadonlySet<string>
+  /** A tool's execute() began during the attempt (tracked by the tool wrapper, not inferred from part status). */
+  toolExecutionStarted?: boolean
+}): Promise<Discard> {
+  if (input.toolExecutionStarted) return { ok: false, dispatched: ["tool execution in flight"] }
+  const added = MessageV2.parts(input.messageID).filter((part) => !input.before.has(part.id))
+  const acted = added.filter(
+    (part) => part.type === "step-finish" || (part.type === "tool" && part.state.status !== "pending"),
+  )
+  if (acted.length > 0) {
+    return { ok: false, dispatched: acted.map((part) => (part.type === "tool" ? part.tool : part.type)) }
+  }
+  // One statement, so a failure leaves the attempt's output intact rather than half deleted.
+  const stale = added.filter((part) => DISCARDED_WHEN_RETRIED.has(part.type))
+  const removed = new Set<string>(stale.map((part) => part.id))
+  if (stale.length > 0) {
+    Database.use((db) => {
+      db.delete(PartTable)
+        .where(
+          and(
+            eq(PartTable.session_id, input.sessionID),
+            inArray(
+              PartTable.id,
+              stale.map((part) => part.id),
+            ),
+          ),
+        )
+        .run()
+      for (const part of stale)
+        Database.effect(() =>
+          Bus.publish(MessageV2.Event.PartRemoved, {
+            sessionID: input.sessionID,
+            messageID: input.messageID,
+            partID: part.id,
+          }),
+        )
+    })
+  }
+  if (removed.size > 0)
+    log.info("discarded partial output before retry", { messageID: input.messageID, parts: removed.size })
+  return { ok: true, removed }
+}
+
+export * as StallRecovery from "./stall-recovery"