From bb150dac1d0c8f010a77e707a34aebe021679b60 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 11:38:04 -0700 Subject: [PATCH 01/11] fix(session): recover from stalled model requests A model request that never answered hung the session until an external kill, an idle-stream timeout ended it as an unknown error, and a stream that died part-way left its partial output in the message on retry. - Arm a 300s first-byte (response headers) timeout by default; it is configurable per provider through `options.headerTimeout`. - Watch Bedrock Converse binary event streams with the idle timeout too, not only SSE. - Classify a header timeout and an idle-stream timeout as retryable `APIError`s that say "The model stopped responding" (same mapping as upstream OpenCode), so the existing backoff retries them. - Before a retry, discard what the failed attempt streamed (text, reasoning, step-start, tool calls still pending). If the attempt already dispatched a tool call, do not retry, so the tool is never re-run; the session ends with an error naming the tool. - Never retry after a user cancel. Closes #1424 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/docs/configure/providers.md | 19 + packages/opencode/src/provider/error.ts | 5 + packages/opencode/src/provider/provider.ts | 20 +- packages/opencode/src/session/message-v2.ts | 26 + packages/opencode/src/session/processor.ts | 44 +- .../opencode/src/session/stall-recovery.ts | 54 ++ .../chunk-timeout-844.test.ts | 6 +- .../test/session/stall-recovery.test.ts | 658 ++++++++++++++++++ 8 files changed, 825 insertions(+), 7 deletions(-) create mode 100644 packages/opencode/src/session/stall-recovery.ts create mode 100644 packages/opencode/test/session/stall-recovery.test.ts diff --git a/docs/docs/configure/providers.md b/docs/docs/configure/providers.md index 38ee4c2dd5..d835ea07cf 100644 --- a/docs/docs/configure/providers.md +++ b/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` disables it | +| `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 } } + } +} +``` diff --git a/packages/opencode/src/provider/error.ts b/packages/opencode/src/provider/error.ts index fd8c849f2c..f81a7a9f4b 100644 --- a/packages/opencode/src/provider/error.ts +++ b/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" diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 3115c1164a..0fbf8b6ba2 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/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..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,7 +149,11 @@ 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") ?? "" + 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({ @@ -149,7 +161,7 @@ export namespace Provider { const part = await new Promise>>((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"] diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 152b75774b..36485bf156 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -1171,6 +1171,32 @@ 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. + 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: + return new MessageV2.APIError( + { + message: + e.message === ProviderError.SSE_IDLE_MESSAGE + ? `${ProviderError.MODEL_STOPPED_RESPONDING}: the response stream went silent (${e.message})` + : 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, diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 81cfd3036f..443050ba9f 100644 --- a/packages/opencode/src/session/processor.ts +++ b/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" @@ -252,7 +256,8 @@ export namespace SessionProcessor { // coerce malformed tool-call ids at ingestion; sanitized ids are used as // BOTH the persisted callID and the pairing key. Salted per processor so // regenerated ids for empty/duplicate raw values cannot collide across steps. - const coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) + // `let`: a retried request starts a fresh id table so no id from the discarded attempt can pair with it. + let coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) const consumeToolCallID = (raw: unknown, identity: ToolCallIdentity) => { // Local tool wrappers preserve the exact execution occurrence through // settlement, even when the provider repeats one malformed raw ID and @@ -408,7 +413,13 @@ 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() + // 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 = {} if (snapshot === undefined) { @@ -1195,8 +1206,24 @@ 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, + }) + : undefined + if (discard && !discard.ok) { + log.warn("not retrying: a tool call was already dispatched", { tools: discard.dispatched }) + } + if (retry !== undefined && discard?.ok) { + for (const [id, part] of toolcalls) if (discard.removed.has(part.id)) toolcalls.delete(id) + coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) // altimate_change end attempt++ const delay = SessionRetry.delay(attempt, error.name === "APIError" ? error : undefined) @@ -1221,6 +1248,17 @@ export namespace SessionProcessor { }) } // altimate_change end + // altimate_change start — say plainly that the model stopped answering and why we gave up + if ( + MessageV2.APIError.isInstance(error) && + error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) + ) { + error.data.message += + discard && !discard.ok + ? ` (not retried: a ${discard.dispatched.join(", ")} tool call had already been dispatched)` + : ` (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 diff --git a/packages/opencode/src/session/stall-recovery.ts b/packages/opencode/src/session/stall-recovery.ts new file mode 100644 index 0000000000..1d63cb38b7 --- /dev/null +++ b/packages/opencode/src/session/stall-recovery.ts @@ -0,0 +1,54 @@ +import { Log } from "@/util/log" +import { MessageV2 } from "./message-v2" +import { Session } from "." +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. + */ +export namespace StallRecovery { + 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 = new Set([ + "text", + "reasoning", + "step-start", + "tool", + ]) + + export function partIDs(messageID: MessageID): Set { + return new Set(MessageV2.parts(messageID).map((part) => part.id)) + } + + export type Discard = { ok: true; removed: Set } | { ok: false; dispatched: string[] } + + export async function discardAttempt(input: { + sessionID: SessionID + messageID: MessageID + before: ReadonlySet + }): Promise { + const added = MessageV2.parts(input.messageID).filter((part) => !input.before.has(part.id)) + const dispatched = added.filter((part) => part.type === "tool" && part.state.status !== "pending") + if (dispatched.length > 0) { + return { ok: false, dispatched: dispatched.map((part) => (part.type === "tool" ? part.tool : part.type)) } + } + const removed = new Set() + for (const part of added) { + if (!DISCARDED_WHEN_RETRIED.has(part.type)) continue + await Session.removePart({ sessionID: input.sessionID, messageID: input.messageID, partID: part.id }) + removed.add(part.id) + } + if (removed.size > 0) + log.info("discarded partial output before retry", { messageID: input.messageID, parts: removed.size }) + return { ok: true, removed } + } +} diff --git a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts index d962718e37..c6aae0bf03 100644 --- a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts +++ b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts @@ -213,7 +213,11 @@ describe("PR #844 — DEFAULT_CHUNK_TIMEOUT 2min→5min", () => { expect(normalized).toContain("function wrapSSE(res: Response, ms: number, ctl: AbortController)") expect(normalized).toContain('if (typeof ms !== "number" || ms <= 0) return res') expect(normalized).toContain("if (!res.body) return res") - expect(normalized).toContain('if (!res.headers.get("content-type")?.includes("text/event-stream")) return res') + // SSE and Bedrock's binary event-stream are watched; every other content type passes through + expect(normalized).toContain('const contentType = res.headers.get("content-type") ?? ""') + expect(normalized).toContain( + '!contentType.includes("text/event-stream") && !contentType.includes("application/vnd.amazon.eventstream")', + ) }) }) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts new file mode 100644 index 0000000000..9afc03c13e --- /dev/null +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -0,0 +1,658 @@ +// Drives the REAL session code (SessionProcessor -> LLM.stream -> Provider fetch wrapper -> AI SDK) +// against a local HTTP server that speaks the OpenAI-compatible streaming protocol and, on demand, +// accepts a request and never answers, streams a few chunks and goes silent, or resets the socket. +import { Database } from "@opencode-ai/core/database/database" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2Bridge } from "@/event-v2-bridge" +import { afterEach, describe, expect, spyOn, test } from "bun:test" +import { jsonSchema, tool } from "ai" +import { Effect, Layer } from "effect" +import net from "net" +import path from "path" +import type { Agent } from "../../src/agent/agent" +import { Provider } from "@/provider/provider" +import { Session } from "@/session/session" +import type { 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, PartID, SessionID } from "../../src/session/schema" +import { ProviderID, ModelID } from "@/provider/schema" +import { ProviderError } from "../../src/provider/error" +import { SessionStatus } from "../../src/session/status" +import { SessionSummary } from "../../src/session/summary" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { testEffect } from "../lib/effect" +import { provideTmpdirInstanceLegacy } from "./legacy-instance" + +const summary = Layer.succeed( + SessionSummary.Service, + SessionSummary.Service.of({ + summarize: () => Effect.void, + diff: () => Effect.succeed([]), + computeDiff: () => Effect.succeed([]), + }), +) +const root = LayerNode.group([ + SessionProcessor.node, + Session.node, + SessionProjector.node, + Provider.node, + Database.node, + EventV2Bridge.node, + SessionStatus.node, + CrossSpawnSpawner.node, +]) +const it = testEffect( + LayerNode.buildLayer(root, { + replacements: [ + LayerNode.replace(SessionSummary.node, summary), + LayerNode.replace(RuntimeFlags.node, RuntimeFlags.layer({ experimentalEventSystem: true })), + ], + }), +) + +// --------------------------------------------------------------------------- +// Fake OpenAI-compatible streaming server +// --------------------------------------------------------------------------- + +type Step = + /** Accept the request and never write a single byte (no headers either). */ + | { kind: "silent" } + /** Optionally wait, answer with SSE headers, emit `events`, then end / go quiet / reset. */ + | { + kind: "stream" + delayMs?: number + gapMs?: number + contentType?: string + events: Array<{ text?: string; tool?: { id: string; name: string; args: string } }> + end: "stop" | "tool_calls" | "silent" | "reset" + } + +type FakeServer = { + url: string + requests: number + /** Sockets the server saw close (client abort, timeout or our own reset). */ + closed: number + bodies: any[] + script: Step[] + stop: () => void +} + +// A raw TCP/HTTP-1.1 server: Bun.serve cannot produce a real mid-body connection reset, and the +// stall shapes below need byte-level control (no headers at all, silent after headers, RST). +function fakeServer(script: Step[]): FakeServer { + const state: FakeServer = { url: "", requests: 0, closed: 0, bodies: [], script, stop: () => {} } + const sockets = new Set() + const frame = (delta: Record, finish: string | null = null) => { + const payload = `data: ${JSON.stringify({ id: "c1", object: "chat.completion.chunk", choices: [{ index: 0, delta, finish_reason: finish }] })}\n\n` + return `${Buffer.byteLength(payload).toString(16)}\r\n${payload}\r\n` + } + const done = () => { + const payload = "data: [DONE]\n\n" + return `${Buffer.byteLength(payload).toString(16)}\r\n${payload}\r\n0\r\n\r\n` + } + const respond = async (sock: net.Socket, step: Step) => { + if (step.kind === "silent") return // read the request, write nothing, keep the socket open + if (step.delayMs) await Bun.sleep(step.delayMs) + if (sock.destroyed) return + sock.write( + `HTTP/1.1 200 OK\r\ncontent-type: ${step.contentType ?? "text/event-stream"}\r\ntransfer-encoding: chunked\r\nconnection: close\r\n\r\n`, + ) + sock.write(frame({ role: "assistant" })) + for (const ev of step.events) { + if (step.gapMs) await Bun.sleep(step.gapMs) + if (sock.destroyed) return + if (ev.text) sock.write(frame({ content: ev.text })) + if (ev.tool) + sock.write( + frame({ + tool_calls: [ + { index: 0, id: ev.tool.id, type: "function", function: { name: ev.tool.name, arguments: ev.tool.args } }, + ], + }), + ) + } + if (step.end === "stop" || step.end === "tool_calls") { + sock.write(frame({}, step.end)) + sock.write(done()) + sock.end() + } + if (step.end === "reset") { + await Bun.sleep(150) // let the partial chunks reach the client before the socket dies + sock.resetAndDestroy() + } + // "silent": leave the stream open and write nothing more. + } + const server = net.createServer((sock) => { + sockets.add(sock) + sock.on("error", () => {}) + sock.on("close", () => { + sockets.delete(sock) + state.closed++ + }) + let buf = Buffer.alloc(0) + let handled = false + sock.on("data", (data) => { + if (handled) return + buf = Buffer.concat([buf, data]) + const end = buf.indexOf("\r\n\r\n") + if (end < 0) return + const head = buf.subarray(0, end).toString() + const length = Number(/content-length:\s*(\d+)/i.exec(head)?.[1] ?? 0) + if (buf.length < end + 4 + length) return + handled = true + state.bodies.push(JSON.parse(buf.subarray(end + 4, end + 4 + length).toString() || "null")) + const index = state.requests++ + // Past the end of the script: behave like a healthy provider so a runaway retry loop shows up as a count. + const step = state.script[index] ?? ({ kind: "stream", events: [{ text: "ok" }], end: "stop" } as Step) + void respond(sock, step) + }) + }) + server.listen(0, "127.0.0.1") + state.url = `http://127.0.0.1:${(server.address() as net.AddressInfo).port}/v1` + state.stop = () => { + for (const s of sockets) s.destroy() + server.close() + } + return state +} + +// --------------------------------------------------------------------------- +// Session harness +// --------------------------------------------------------------------------- + +const ref = { providerID: "test", modelID: "test-model" } + +function providerCfg(url: string, options: Record = {}) { + return { + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { apiKey: "test-key", baseURL: url, ...options }, + }, + }, + } +} + +const agent = (): Agent.Info => ({ + name: "build", + mode: "primary", + options: {}, + permission: [{ permission: "*", pattern: "*", action: "allow" }], +}) + +type RunOptions = { + script: Step[] + /** Provider options (chunkTimeout / headerTimeout): the existing per-provider configuration surface. */ + options?: Record + tools?: Record + /** Abort the user signal this long after the request count first reaches 1. */ + cancelAfterMs?: number +} + +const servers: FakeServer[] = [] +afterEach(() => { + for (const s of servers.splice(0)) s.stop() +}) + +function runSession(opts: RunOptions) { + const server = fakeServer(opts.script) + servers.push(server) + return provideTmpdirInstanceLegacy( + (dir) => + Effect.gen(function* () { + const processors = yield* SessionProcessor.Service + const session = yield* Session.Service + const provider = yield* Provider.Service + + const chat = yield* session.create({}) + const parent = yield* session.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID: chat.id, + agent: "build", + model: ref as any, + time: { created: Date.now() }, + }) + yield* session.updatePart({ + id: PartID.ascending(), + messageID: parent.id, + sessionID: chat.id, + type: "text", + text: "hi", + }) + const msg = { + id: MessageID.ascending(), + role: "assistant", + sessionID: chat.id, + mode: "build", + agent: "build", + path: { cwd: dir, root: dir }, + cost: 0, + tokens: { total: 0, input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: ref.modelID, + providerID: ref.providerID, + parentID: parent.id, + time: { created: Date.now() }, + finish: "end_turn", + } as unknown as MessageV2.Assistant + yield* session.updateMessage(msg as any) + const mdl = yield* provider.getModel(ProviderID.make(ref.providerID), ModelID.make(ref.modelID)) + const controller = new AbortController() + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + abort: controller.signal, + }) + const started = Date.now() + if (opts.cancelAfterMs !== undefined) { + const poll = setInterval(() => { + if (server.requests > 0) { + clearInterval(poll) + setTimeout(() => controller.abort(), opts.cancelAfterMs) + } + }, 5) + } + const result = yield* Effect.promise(() => + handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: "build", + model: ref, + } as unknown as MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "hi" }], + tools: opts.tools ?? {}, + abort: controller.signal, + } as LLM.StreamInput), + ) + const elapsed = Date.now() - started + // Let the processor's trailing fire-and-forget work drain; closing the scope under it makes bun + // report "All fibers interrupted" against whichever test ran last (also seen in processor-effect.test.ts). + yield* Effect.sleep("300 millis") + const parts = MessageV2.parts(msg.id) + return { result, parts, message: handle.message, server, elapsed } + }), + { config: () => providerCfg(server.url, opts.options) as any }, + ) +} + +const text = (parts: MessageV2.Part[]) => + parts + .filter((p): p is MessageV2.TextPart => p.type === "text") + .map((p) => p.text) + .join("") + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +const ok = (t: string): Step => ({ kind: "stream", events: [{ text: t }], end: "stop" }) +const echoTool = (calls: string[]) => ({ + echo: tool({ + description: "echo", + inputSchema: jsonSchema<{ cmd: string }>({ + type: "object", + properties: { cmd: { type: "string" } }, + required: ["cmd"], + }), + execute: async ({ cmd }) => { + calls.push(cmd) + return { title: "echo", output: `ran ${cmd}`, metadata: {} } + }, + }), +}) +const countOf = (parts: MessageV2.Part[], type: MessageV2.Part["type"]) => parts.filter((p) => p.type === type).length + +// Backoff is irrelevant to these assertions; keep the suite fast. +let backoff: ReturnType | undefined +const fastBackoff = (ms = 5) => (backoff = spyOn(SessionRetry, "delay").mockReturnValue(ms)) +afterEach(() => backoff?.mockRestore()) + +// Short windows so a stall is noticed in well under a second; healthy-slow tests use wider ones. +const FAST = { headerTimeout: 400, chunkTimeout: 400 } + +it.live("request is accepted but never answered: first-byte timeout aborts it and the retry succeeds", () => { + fastBackoff() + return runSession({ script: [{ kind: "silent" }, ok("hello")], options: FAST }).pipe( + Effect.tap(({ result, parts, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(2) + expect(result).toBe("continue") + expect(message.error).toBeUndefined() + expect(text(parts)).toBe("hello") + // the stalled connection was really closed, not merely abandoned + expect(server.closed).toBeGreaterThanOrEqual(1) + }), + ), + ) +}) + +it.live("goes quiet mid-stream: idle timeout aborts, partial text is discarded, the retry succeeds", () => { + fastBackoff() + return runSession({ + script: [ + { kind: "stream", events: [{ text: "Let me " }, { text: "think about " }], end: "silent" }, + ok("final answer"), + ], + options: FAST, + }).pipe( + Effect.tap(({ result, parts, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(2) + expect(result).toBe("continue") + expect(message.error).toBeUndefined() + expect(text(parts)).toBe("final answer") + // nothing from the stalled attempt is left behind: one step, one text part + expect(countOf(parts, "text")).toBe(1) + expect(countOf(parts, "step-start")).toBe(1) + expect(server.closed).toBeGreaterThanOrEqual(1) + }), + ), + ) +}) + +it.live("binary event-stream (Bedrock Converse) responses get the same idle timeout", () => { + fastBackoff() + return runSession({ + script: [ + { kind: "stream", contentType: "application/vnd.amazon.eventstream", events: [{ text: "x" }], end: "silent" }, + ok("done"), + ], + options: FAST, + }).pipe( + Effect.tap(({ result, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(2) + expect(result).toBe("continue") + expect(message.error).toBeUndefined() + }), + ), + ) +}) + +it.live("connection reset mid-stream: partial text is discarded and the retry succeeds", () => { + fastBackoff() + return runSession({ + script: [{ kind: "stream", events: [{ text: "partial " }], end: "reset" }, ok("complete")], + options: FAST, + }).pipe( + Effect.tap(({ result, parts, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(2) + expect(result).toBe("continue") + expect(message.error).toBeUndefined() + expect(text(parts)).toBe("complete") + expect(countOf(parts, "text")).toBe(1) + expect(countOf(parts, "step-start")).toBe(1) + }), + ), + ) +}) + +it.live("a slow but healthy stream is NOT cut off (late first byte, slow steady chunks)", () => + runSession({ + script: [ + { + kind: "stream", + delayMs: 600, // first byte well after most of the window, still inside it + gapMs: 400, // each gap inside the idle window; the whole stream outlasts it + events: [{ text: "a" }, { text: "b" }, { text: "c" }, { text: "d" }, { text: "e" }], + end: "stop", + }, + ], + options: { headerTimeout: 1500, chunkTimeout: 1000 }, + }).pipe( + Effect.tap(({ result, parts, message, server, elapsed }) => + Effect.sync(() => { + expect(elapsed).toBeGreaterThan(1500) // outlived both windows, so the test is not vacuous + expect(server.requests).toBe(1) + expect(result).toBe("continue") + expect(message.error).toBeUndefined() + expect(text(parts)).toBe("abcde") + }), + ), + ), +) + +it.live("partially streamed tool call (input still pending) is discarded and the tool runs exactly once", () => { + fastBackoff() + const calls: string[] = [] + return runSession({ + tools: echoTool(calls), + script: [ + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"ec' } }], end: "silent" }, + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"go"}' } }], end: "tool_calls" }, + ], + options: FAST, + }).pipe( + Effect.tap(({ parts, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(2) + expect(message.error).toBeUndefined() + expect(calls).toEqual(["go"]) + const toolParts = parts.filter((p): p is MessageV2.ToolPart => p.type === "tool") + expect(toolParts).toHaveLength(1) + expect(toolParts[0].state.status).toBe("completed") + }), + ), + ) +}) + +it.live("a tool call already dispatched is never re-run: the stall is reported instead of retried", () => { + fastBackoff() + const calls: string[] = [] + return runSession({ + tools: echoTool(calls), + script: [ + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"rm"}' } }], end: "silent" }, + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"rm"}' } }], end: "tool_calls" }, + ], + options: FAST, + }).pipe( + Effect.tap(({ result, message, server }) => + Effect.sync(() => { + expect(calls).toEqual(["rm"]) // ran once, and only once + expect(server.requests).toBe(1) // no second request that would re-emit the call + expect(result).toBe("stop") + expect(message.error?.name).toBe("APIError") + const msg = (message.error?.data as { message: string }).message + expect(msg).toContain("The model stopped responding") + expect(msg).toContain("not retried") + expect(msg).toContain("echo") + }), + ), + ) +}) + +it.live("retries are bounded and exhaustion ends the session with a clear 'model stopped responding' error", () => { + fastBackoff() + return runSession({ script: Array.from({ length: 20 }, () => ({ kind: "silent" }) as Step), options: FAST }).pipe( + Effect.tap(({ result, message, server }) => + Effect.sync(() => { + expect(server.requests).toBe(1 + SessionRetry.RETRY_MAX_ATTEMPTS) + expect(result).toBe("stop") + expect(message.error?.name).toBe("APIError") + const msg = (message.error?.data as { message: string }).message + expect(msg).toContain("The model stopped responding") + expect(msg).toContain(`gave up after ${SessionRetry.RETRY_MAX_ATTEMPTS} retries`) + }), + ), + ) +}) + +it.live("user cancel during the first-byte wait ends the session at once and is not treated as a stall", () => { + fastBackoff() + return runSession({ + script: [{ kind: "silent" }, ok("never reached")], + options: { headerTimeout: 10_000, chunkTimeout: 10_000 }, + cancelAfterMs: 100, + }).pipe( + Effect.tap(({ message, server, elapsed }) => + Effect.sync(() => { + expect(elapsed).toBeLessThan(3000) // far below the 10s windows: the abort did it + expect(server.requests).toBe(1) + expect(message.error?.name).toBe("MessageAbortedError") + expect(server.closed).toBeGreaterThanOrEqual(1) + }), + ), + ) +}) + +it.live("user cancel during the retry backoff ends the session at once and sends no further request", () => { + fastBackoff(30_000) + return runSession({ + script: [{ kind: "silent" }, ok("never reached")], + options: FAST, + cancelAfterMs: 800, // first-byte timeout fires at ~400ms; the session is then sleeping in backoff + }).pipe( + Effect.tap(({ message, server, elapsed }) => + Effect.sync(() => { + expect(elapsed).toBeLessThan(5000) + expect(server.requests).toBe(1) + expect(message.error?.name).toBe("MessageAbortedError") + }), + ), + ) +}) + +it.live("every stall timer is cleared: none outlive the request, stalled or healthy", () => { + fastBackoff() + // Unusual durations identify this suite's own watchdog timers among everything else using setTimeout. + const HEADER = 777 + const CHUNK = 888 + const realSet = globalThis.setTimeout + const realClear = globalThis.clearTimeout + const live = new Set() + let created = 0 + globalThis.setTimeout = ((fn: any, ms?: number, ...rest: any[]) => { + if (ms !== HEADER && ms !== CHUNK) return realSet(fn, ms, ...rest) + created++ + const id: unknown = realSet( + (...a: any[]) => { + live.delete(id) + return fn(...a) + }, + ms, + ...rest, + ) + live.add(id) + return id + }) as typeof setTimeout + globalThis.clearTimeout = ((id: any) => { + live.delete(id) + return realClear(id) + }) as typeof clearTimeout + const restore = () => { + globalThis.setTimeout = realSet + globalThis.clearTimeout = realClear + } + return runSession({ + // one stalled attempt (timer fires), one healthy multi-chunk attempt (timers cleared as chunks arrive) + script: [{ kind: "silent" }, { kind: "stream", events: [{ text: "a" }, { text: "b" }], end: "stop" }], + options: { headerTimeout: HEADER, chunkTimeout: CHUNK }, + }).pipe( + Effect.tap(({ message }) => + Effect.sync(() => { + expect(message.error).toBeUndefined() + expect(created).toBeGreaterThan(2) + expect(live.size).toBe(0) + }), + ), + Effect.ensuring(Effect.sync(restore)), + ) +}) + +it.live("with no headerTimeout configured a 300s first-byte timer is armed and cleared once headers arrive", () => { + const DEFAULT = 300_000 + const realSet = globalThis.setTimeout + const realClear = globalThis.clearTimeout + const live = new Set() + let created = 0 + globalThis.setTimeout = ((fn: any, ms?: number, ...rest: any[]) => { + const id = realSet(fn, ms, ...rest) + if (ms === DEFAULT) { + created++ + live.add(id) + } + return id + }) as typeof setTimeout + globalThis.clearTimeout = ((id: any) => { + live.delete(id) + return realClear(id) + }) as typeof clearTimeout + return runSession({ script: [ok("hi")], options: { chunkTimeout: 555 } }).pipe( + Effect.tap(({ message }) => + Effect.sync(() => { + expect(message.error).toBeUndefined() + expect(created).toBe(1) // the default first-byte timer, nothing else uses 300s here + expect(live.size).toBe(0) + }), + ), + Effect.ensuring( + Effect.sync(() => { + globalThis.setTimeout = realSet + globalThis.clearTimeout = realClear + }), + ), + ) +}) + +describe("stall classification and defaults", () => { + const providerID = ProviderID.make("test") + + test("a header timeout is a retryable APIError that says the model stopped responding", () => { + const error = MessageV2.fromError(new ProviderError.HeaderTimeoutError(120_000), { providerID }) + expect(error.name).toBe("APIError") + expect((error.data as any).message).toBe("The model stopped responding: no response headers within 120s") + expect(SessionRetry.retryable(error)).toContain("The model stopped responding") + }) + + test("an idle-stream timeout is a retryable APIError that says the model stopped responding", () => { + const error = MessageV2.fromError(new ProviderError.ResponseStreamError(ProviderError.SSE_IDLE_MESSAGE), { + providerID, + }) + expect(error.name).toBe("APIError") + expect((error.data as any).message).toContain("The model stopped responding") + expect(SessionRetry.retryable(error)).toBeDefined() + }) + + test("other stream failures keep their own message but are retryable", () => { + const error = MessageV2.fromError(new ProviderError.ResponseStreamError("WebSocket closed"), { providerID }) + expect((error.data as any).message).toBe("WebSocket closed") + expect(SessionRetry.retryable(error)).toBeDefined() + }) + + test("a user abort is still an abort, not a stall", () => { + const error = MessageV2.fromError(new DOMException("Aborted", "AbortError"), { providerID }) + expect(error.name).toBe("MessageAbortedError") + expect(SessionRetry.retryable(error)).toBeUndefined() + }) +}) From a0e99ddc5bbdb1cba978a4b33411b4117c30830f Mon Sep 17 00:00:00 2001 From: Anand Gupta Date: Wed, 7 Oct 2026 12:03:19 -0700 Subject: [PATCH 02/11] fix(session): harden stall retry after review - Refuse a retry once the failed attempt finished its step, so cost and tokens are not counted twice and the answer is not produced twice. - Explain on any `APIError` why a retry was refused, and stop logging that case as exhausted retries. - Keep a failed partial-output cleanup from escaping the catch block. - Do not append "gave up after N retries" when the user cancelled. - Test the discard rule directly. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/session/processor.ts | 28 +++--- .../opencode/src/session/stall-recovery.ts | 10 ++- .../test/session/stall-recovery.test.ts | 87 +++++++++++++++++++ 3 files changed, 111 insertions(+), 14 deletions(-) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 443050ba9f..a53db3f5be 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -1216,10 +1216,14 @@ export namespace SessionProcessor { sessionID: input.sessionID, messageID: input.assistantMessage.id, before: partsBeforeAttempt, + }).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: a tool call was already dispatched", { tools: discard.dispatched }) + 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) @@ -1239,7 +1243,7 @@ export namespace SessionProcessor { continue } // altimate_change start — log when retries exhausted for debugging - if (retry !== undefined) { + if (retry !== undefined && !(discard && !discard.ok)) { log.warn("max retry attempts reached, giving up", { attempt, message: retry, @@ -1248,15 +1252,17 @@ export namespace SessionProcessor { }) } // altimate_change end - // altimate_change start — say plainly that the model stopped answering and why we gave up - if ( - MessageV2.APIError.isInstance(error) && - error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) - ) { - error.data.message += - discard && !discard.ok - ? ` (not retried: a ${discard.dispatched.join(", ")} tool call had already been dispatched)` - : ` (gave up after ${attempt} retries)` + // 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 ( + retry !== undefined && + !input.abort.aborted && + error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) + ) { + error.data.message += ` (gave up after ${attempt} retries)` + } } // altimate_change end input.assistantMessage.error = error diff --git a/packages/opencode/src/session/stall-recovery.ts b/packages/opencode/src/session/stall-recovery.ts index 1d63cb38b7..9e755a3179 100644 --- a/packages/opencode/src/session/stall-recovery.ts +++ b/packages/opencode/src/session/stall-recovery.ts @@ -13,6 +13,8 @@ import type { MessageID, SessionID } from "./schema" * - 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. */ export namespace StallRecovery { const log = Log.create({ service: "session.stall-recovery" }) @@ -37,9 +39,11 @@ export namespace StallRecovery { before: ReadonlySet }): Promise { const added = MessageV2.parts(input.messageID).filter((part) => !input.before.has(part.id)) - const dispatched = added.filter((part) => part.type === "tool" && part.state.status !== "pending") - if (dispatched.length > 0) { - return { ok: false, dispatched: dispatched.map((part) => (part.type === "tool" ? part.tool : part.type)) } + 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)) } } const removed = new Set() for (const part of added) { diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 9afc03c13e..ea9de93ce3 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -15,6 +15,7 @@ import { Session } from "@/session/session" import type { LLM } from "../../src/session/llm" import { MessageV2 } from "../../src/session/message-v2" import { SessionProcessor } from "../../src/session/processor" +import { StallRecovery } from "../../src/session/stall-recovery" import { SessionRetry } from "../../src/session/retry" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { ProviderID, ModelID } from "@/provider/schema" @@ -625,6 +626,92 @@ it.live("with no headerTimeout configured a 300s first-byte timer is armed and c ) }) +// Direct tests of the discard rule on a hand-built message (the SDK cannot be made to emit these orderings). +const seed = (parts: Array>) => + provideTmpdirInstanceLegacy((dir) => + Effect.gen(function* () { + const session = yield* Session.Service + const chat = yield* session.create({}) + const messageID = MessageID.ascending() + yield* session.updateMessage({ + id: messageID, + role: "assistant", + sessionID: chat.id, + mode: "build", + agent: "build", + path: { cwd: dir, root: dir }, + cost: 0, + tokens: { total: 0, input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: ref.modelID, + providerID: ref.providerID, + parentID: MessageID.ascending(), + time: { created: Date.now() }, + finish: "end_turn", + } as any) + const before = StallRecovery.partIDs(messageID) + for (const part of parts) + yield* session.updatePart({ id: PartID.ascending(), messageID, sessionID: chat.id, ...part } as any) + const result = yield* Effect.promise(() => + StallRecovery.discardAttempt({ sessionID: chat.id, messageID, before }), + ) + return { result, left: MessageV2.parts(messageID).map((p) => p.type) } + }), + ) +const pendingTool = { type: "tool", tool: "echo", callID: "c1", state: { status: "pending", input: {}, raw: "" } } +const runningTool = { + type: "tool", + tool: "echo", + callID: "c2", + state: { status: "running", input: {}, time: { start: 1 } }, +} + +it.live("discard removes streamed text, reasoning, step-start and pending tool input", () => + seed([ + { type: "step-start" }, + { type: "reasoning", text: "r", time: { start: 1 } }, + { type: "text", text: "t" }, + pendingTool, + ]).pipe( + Effect.tap(({ result, left }) => + Effect.sync(() => { + expect(result.ok).toBe(true) + expect(left).toEqual([]) + }), + ), + ), +) + +it.live("discard refuses, and removes nothing, once a tool call has started", () => + seed([{ type: "text", text: "t" }, pendingTool, runningTool]).pipe( + Effect.tap(({ result, left }) => + Effect.sync(() => { + expect(result).toEqual({ ok: false, dispatched: ["echo"] }) + expect(left).toEqual(["text", "tool", "tool"]) + }), + ), + ), +) + +it.live("discard refuses once the attempt has finished its step (no double-counted cost)", () => + seed([ + { type: "step-start" }, + { type: "text", text: "t" }, + { + type: "step-finish", + reason: "stop", + cost: 1, + tokens: { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } }, + }, + ]).pipe( + Effect.tap(({ result, left }) => + Effect.sync(() => { + expect(result).toEqual({ ok: false, dispatched: ["step-finish"] }) + expect(left).toEqual(["step-start", "text", "step-finish"]) + }), + ), + ), +) + describe("stall classification and defaults", () => { const providerID = ProviderID.make("test") From b2af9321449dd0c2905a8be67e08ad0e87ba076f Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 12:14:59 -0700 Subject: [PATCH 03/11] fix(session): address review feedback on stall retry - Use flat exports with a self re-export in `stall-recovery.ts`, per the package conventions. - Only the idle watchdog's `ResponseStreamError` is retryable; websocket transport errors keep their previous handling. - Make `SessionRetry.sleep` reject at once on an already-aborted signal so a Stop that lands just before the backoff is not delayed. - Log "max retry attempts reached" and append "gave up after N retries" only when attempts were actually exhausted. - Document that `headerTimeout: false` removes the hang protection. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/docs/configure/providers.md | 2 +- packages/opencode/src/session/message-v2.ts | 10 ++- packages/opencode/src/session/processor.ts | 9 +-- packages/opencode/src/session/retry.ts | 4 ++ .../opencode/src/session/stall-recovery.ts | 63 +++++++++---------- packages/opencode/test/session/retry.test.ts | 12 +++- .../test/session/stall-recovery.test.ts | 6 +- 7 files changed, 53 insertions(+), 53 deletions(-) diff --git a/docs/docs/configure/providers.md b/docs/docs/configure/providers.md index d835ea07cf..10e83d44f6 100644 --- a/docs/docs/configure/providers.md +++ b/docs/docs/configure/providers.md @@ -503,7 +503,7 @@ 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` disables it | +| `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 diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 36485bf156..2f2d54043f 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -1174,7 +1174,8 @@ export namespace MessageV2 { // 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. + // (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( { @@ -1184,13 +1185,10 @@ export namespace MessageV2 { }, { cause: e }, ).toObject() - case e instanceof ProviderError.ResponseStreamError: + case e instanceof ProviderError.ResponseStreamError && e.message === ProviderError.SSE_IDLE_MESSAGE: return new MessageV2.APIError( { - message: - e.message === ProviderError.SSE_IDLE_MESSAGE - ? `${ProviderError.MODEL_STOPPED_RESPONDING}: the response stream went silent (${e.message})` - : e.message, + message: `${ProviderError.MODEL_STOPPED_RESPONDING}: the response stream went silent (${e.message})`, isRetryable: true, metadata: { code: e.name }, }, diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index a53db3f5be..f3331d9aa0 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -1243,7 +1243,8 @@ export namespace SessionProcessor { continue } // altimate_change start — log when retries exhausted for debugging - if (retry !== undefined && !(discard && !discard.ok)) { + const exhausted = retry !== undefined && attempt >= SessionRetry.RETRY_MAX_ATTEMPTS + if (exhausted) { log.warn("max retry attempts reached, giving up", { attempt, message: retry, @@ -1256,11 +1257,7 @@ export namespace SessionProcessor { if (MessageV2.APIError.isInstance(error)) { if (discard && !discard.ok) { error.data.message += ` (not retried: the attempt had already acted: ${discard.dispatched.join(", ")})` - } else if ( - retry !== undefined && - !input.abort.aborted && - error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) - ) { + } else if (exhausted && error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING)) { error.data.message += ` (gave up after ${attempt} retries)` } } diff --git a/packages/opencode/src/session/retry.ts b/packages/opencode/src/session/retry.ts index 67d1477c34..73000aecbd 100644 --- a/packages/opencode/src/session/retry.ts +++ b/packages/opencode/src/session/retry.ts @@ -13,6 +13,10 @@ export namespace SessionRetry { export async function sleep(ms: number, signal: AbortSignal): Promise { 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")) diff --git a/packages/opencode/src/session/stall-recovery.ts b/packages/opencode/src/session/stall-recovery.ts index 9e755a3179..a2577ac2c6 100644 --- a/packages/opencode/src/session/stall-recovery.ts +++ b/packages/opencode/src/session/stall-recovery.ts @@ -16,43 +16,38 @@ import type { MessageID, SessionID } from "./schema" * - 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. */ -export namespace StallRecovery { - const log = Log.create({ service: "session.stall-recovery" }) +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 = new Set([ - "text", - "reasoning", - "step-start", - "tool", - ]) +// A `tool` part reaching this set is necessarily still `pending` (started ones are rejected first). +const DISCARDED_WHEN_RETRIED: ReadonlySet = new Set(["text", "reasoning", "step-start", "tool"]) - export function partIDs(messageID: MessageID): Set { - return new Set(MessageV2.parts(messageID).map((part) => part.id)) - } +export function partIDs(messageID: MessageID): Set { + return new Set(MessageV2.parts(messageID).map((part) => part.id)) +} - export type Discard = { ok: true; removed: Set } | { ok: false; dispatched: string[] } +export type Discard = { ok: true; removed: Set } | { ok: false; dispatched: string[] } - export async function discardAttempt(input: { - sessionID: SessionID - messageID: MessageID - before: ReadonlySet - }): Promise { - 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)) } - } - const removed = new Set() - for (const part of added) { - if (!DISCARDED_WHEN_RETRIED.has(part.type)) continue - await Session.removePart({ sessionID: input.sessionID, messageID: input.messageID, partID: part.id }) - removed.add(part.id) - } - if (removed.size > 0) - log.info("discarded partial output before retry", { messageID: input.messageID, parts: removed.size }) - return { ok: true, removed } +export async function discardAttempt(input: { + sessionID: SessionID + messageID: MessageID + before: ReadonlySet +}): Promise { + 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)) } } + const removed = new Set() + for (const part of added) { + if (!DISCARDED_WHEN_RETRIED.has(part.type)) continue + await Session.removePart({ sessionID: input.sessionID, messageID: input.messageID, partID: part.id }) + removed.add(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" diff --git a/packages/opencode/test/session/retry.test.ts b/packages/opencode/test/session/retry.test.ts index 27df23206e..8cf97df922 100644 --- a/packages/opencode/test/session/retry.test.ts +++ b/packages/opencode/test/session/retry.test.ts @@ -88,6 +88,14 @@ describe("session.retry.delay", () => { process.emitWarning = originalWarn expect(warnings.some((w) => w.includes("TimeoutOverflowWarning"))).toBe(false) }) + + test("sleep rejects at once when the signal is already aborted (a Stop that landed before the backoff)", async () => { + const controller = new AbortController() + controller.abort() + const started = Date.now() + await expect(SessionRetry.sleep(60_000, controller.signal)).rejects.toThrow("Aborted") + expect(Date.now() - started).toBeLessThan(1000) + }) }) describe("session.retry.max_attempts", () => { @@ -192,9 +200,7 @@ describe("session.message-v2.fromError", () => { // Use a raw TCP server that sends a partial HTTP response then // destroys the socket, triggering an immediate ECONNRESET on the client. const server = createServer((socket) => { - socket.write( - "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n6\r\nHello,\r\n", - ) + socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n6\r\nHello,\r\n") setTimeout(() => socket.destroy(), 20) }) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index ea9de93ce3..3f84ae31fa 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -731,10 +731,10 @@ describe("stall classification and defaults", () => { expect(SessionRetry.retryable(error)).toBeDefined() }) - test("other stream failures keep their own message but are retryable", () => { + test("other stream failures (websocket transport) keep their previous, non-retried handling", () => { const error = MessageV2.fromError(new ProviderError.ResponseStreamError("WebSocket closed"), { providerID }) - expect((error.data as any).message).toBe("WebSocket closed") - expect(SessionRetry.retryable(error)).toBeDefined() + expect(error.name).toBe("UnknownError") + expect(SessionRetry.retryable(error)).toBeUndefined() }) test("a user abort is still an abort, not a stall", () => { From 7372727771d2e469a2a3b212f16ac78ccbbe60bd Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 12:20:54 -0700 Subject: [PATCH 04/11] fix(session): keep tool-call ids unique across a discarded attempt Replace the coercer reset on retry with `discardUnstarted()`, which drops only tool inputs that started streaming but were never called. Ids already allocated (including by earlier calls in the same message) stay reserved, so a retry that reuses a raw provider id can neither pair with the discarded start nor collide with an earlier call. Also make the stall tests deterministic: disable snapshot tracking in the test config (its background git work was what leaked an "All fibers interrupted" error at scope close) instead of sleeping. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/session/processor.ts | 10 +++++--- .../test/session/stall-recovery.test.ts | 23 ++++++++++++++++--- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index f3331d9aa0..5805b2bb8e 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -237,6 +237,11 @@ 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), so a retry that + // reuses the same raw id pairs with its own start. Allocated ids stay reserved, so none is ever reused. + discardUnstarted() { + started.clear() + }, }) } // altimate_change end @@ -256,8 +261,7 @@ export namespace SessionProcessor { // coerce malformed tool-call ids at ingestion; sanitized ids are used as // BOTH the persisted callID and the pairing key. Salted per processor so // regenerated ids for empty/duplicate raw values cannot collide across steps. - // `let`: a retried request starts a fresh id table so no id from the discarded attempt can pair with it. - let coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) + const coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) const consumeToolCallID = (raw: unknown, identity: ToolCallIdentity) => { // Local tool wrappers preserve the exact execution occurrence through // settlement, even when the provider repeats one malformed raw ID and @@ -1227,7 +1231,7 @@ export namespace SessionProcessor { } if (retry !== undefined && discard?.ok) { for (const [id, part] of toolcalls) if (discard.removed.has(part.id)) toolcalls.delete(id) - coerceToolCallID = createToolCallIDCoercer(input.assistantMessage.id) + coerceToolCallID.discardUnstarted() // altimate_change end attempt++ const delay = SessionRetry.delay(attempt, error.name === "APIError" ? error : undefined) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 3f84ae31fa..25c7d8eedd 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -169,6 +169,9 @@ const ref = { providerID: "test", modelID: "test-model" } function providerCfg(url: string, options: Record = {}) { return { + // Snapshot tracking spawns git in the background; closing the test scope under it surfaces as an + // "All fibers interrupted" error on the last test, and these tests have no use for snapshots. + snapshot: false, provider: { test: { name: "Test", @@ -295,9 +298,6 @@ function runSession(opts: RunOptions) { } as LLM.StreamInput), ) const elapsed = Date.now() - started - // Let the processor's trailing fire-and-forget work drain; closing the scope under it makes bun - // report "All fibers interrupted" against whichever test ran last (also seen in processor-effect.test.ts). - yield* Effect.sleep("300 millis") const parts = MessageV2.parts(msg.id) return { result, parts, message: handle.message, server, elapsed } }), @@ -712,6 +712,23 @@ it.live("discard refuses once the attempt has finished its step (no double-count ), ) +describe("tool-call id bookkeeping across a discarded attempt", () => { + test("a retry that reuses the raw id pairs with its own start and never reuses an allocated id", () => { + const ids = SessionProcessor.createToolCallIDCoercer("msg") + // an earlier call in the same message completed under raw id "call_1" + const first = ids.start("call_1") + ids.call("call_1") + ids.result("call_1", first) + // the stalled attempt started the same raw id, never called it, and was discarded + const stale = ids.start("call_1") + ids.discardUnstarted() + // the retry reuses the raw id + const retried = ids.start("call_1") + expect(ids.call("call_1")).toBe(retried) + expect(new Set([first, stale, retried]).size).toBe(3) + }) +}) + describe("stall classification and defaults", () => { const providerID = ProviderID.make("test") From 689b47e8548c828db624ca6edad284af28685b82 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 12:34:45 -0700 Subject: [PATCH 05/11] fix(session): realign tool-call id ordinals after a discarded attempt `discardUnstarted()` now removes the discarded attempt's allocations, so the per-id slot ordinals that `settled()` and `executionID()` index by stay aligned with what actually executes, while ids of earlier calls remain reserved. Tests cover both the collision and the ordinal alignment cases. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/session/processor.ts | 15 +++++++++++++-- .../test/session/stall-recovery.test.ts | 17 +++++++++++++++-- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 5805b2bb8e..5658e2d930 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -237,9 +237,20 @@ 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), so a retry that - // reuses the same raw id pairs with its own start. Allocated ids stay reserved, so none is ever reused. + // 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() }, }) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 25c7d8eedd..20b5c7ccf5 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -713,7 +713,7 @@ it.live("discard refuses once the attempt has finished its step (no double-count ) describe("tool-call id bookkeeping across a discarded attempt", () => { - test("a retry that reuses the raw id pairs with its own start and never reuses an allocated id", () => { + test("a retry reusing the raw id pairs with its own start and never collides with an earlier call", () => { const ids = SessionProcessor.createToolCallIDCoercer("msg") // an earlier call in the same message completed under raw id "call_1" const first = ids.start("call_1") @@ -725,7 +725,20 @@ describe("tool-call id bookkeeping across a discarded attempt", () => { // the retry reuses the raw id const retried = ids.start("call_1") expect(ids.call("call_1")).toBe(retried) - expect(new Set([first, stale, retried]).size).toBe(3) + expect(retried).not.toBe(first) + expect(stale).not.toBe(first) + }) + + test("execution ordinals stay aligned: the retried call resolves to its own id, not the discarded one", () => { + const ids = SessionProcessor.createToolCallIDCoercer("msg") + ids.start("") // empty raw id: repeats across attempts + ids.discardUnstarted() + const retried = ids.start("") + ids.call("") + const execution = ids.beginExecution("") + expect(ids.executionID(execution)).toBe(retried) + ids.finishExecution(execution) + expect(ids.settled("")).toBe(retried) }) }) From aa0d03106c19f62bdac07fcc1bd751643e0c88f4 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 12:58:40 -0700 Subject: [PATCH 06/11] fix(session): keep tool history of a refused-retry stall in replay - Replay an assistant step that ended with "The model stopped responding" and already ran tools, as an aborted step is, so the next turn knows those side effects happened instead of losing them from context. - Match the Bedrock event-stream content type case-insensitively. - Show "Model stopped responding" in the TUI for the retried-out error too. - Tighten the stall tests: assert the stalled request's own socket closed, start the backoff cancel only after the stall fired, clear test timers on every exit, and cover parts that predate the failed attempt. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/provider/provider.ts | 2 +- packages/opencode/src/session/message-v2.ts | 7 +- .../chunk-timeout-844.test.ts | 2 +- .../opencode/test/session/message-v2.test.ts | 40 +++++++++ .../test/session/stall-recovery.test.ts | 81 ++++++++++++++++--- .../feature-plugins/system/notifications.ts | 8 +- .../test/cli/cmd/tui/notifications.test.ts | 31 +++++++ 7 files changed, 153 insertions(+), 18 deletions(-) diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 0fbf8b6ba2..88fc1f0eac 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -150,7 +150,7 @@ export namespace Provider { if (typeof ms !== "number" || ms <= 0) return res if (!res.body) return res // altimate_change start — Bedrock Converse streams binary event-stream frames, not SSE; watch those too - const contentType = res.headers.get("content-type") ?? "" + 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 diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 2f2d54043f..51a100adb8 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -763,15 +763,20 @@ export namespace MessageV2 { const differentModel = `${model.providerID}/${model.id}` !== `${msg.info.providerID}/${msg.info.modelID}` const media: Array<{ mime: string; url: string }> = [] + // altimate_change start — a stalled step whose retry was refused because a tool had already run keeps its + // tool history in replay (as an aborted step does), so the next turn knows those side effects happened if ( msg.info.error && !( - MessageV2.AbortedError.isInstance(msg.info.error) && + (MessageV2.AbortedError.isInstance(msg.info.error) || + (MessageV2.APIError.isInstance(msg.info.error) && + msg.info.error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING))) && msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning") ) ) { continue } + // altimate_change end const assistantMessage: UIMessage = { id: msg.info.id, role: "assistant", diff --git a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts index c6aae0bf03..239024e318 100644 --- a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts +++ b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts @@ -214,7 +214,7 @@ describe("PR #844 — DEFAULT_CHUNK_TIMEOUT 2min→5min", () => { expect(normalized).toContain('if (typeof ms !== "number" || ms <= 0) return res') expect(normalized).toContain("if (!res.body) return res") // SSE and Bedrock's binary event-stream are watched; every other content type passes through - expect(normalized).toContain('const contentType = res.headers.get("content-type") ?? ""') + expect(normalized).toContain('const contentType = (res.headers.get("content-type") ?? "").toLowerCase()') expect(normalized).toContain( '!contentType.includes("text/event-stream") && !contentType.includes("application/vnd.amazon.eventstream")', ) diff --git a/packages/opencode/test/session/message-v2.test.ts b/packages/opencode/test/session/message-v2.test.ts index ca87e6fb29..ef28db241c 100644 --- a/packages/opencode/test/session/message-v2.test.ts +++ b/packages/opencode/test/session/message-v2.test.ts @@ -1054,6 +1054,46 @@ describe("session.message-v2.toModelMessage", () => { ]) }) + test("keeps the tool history of a stalled step whose retry was refused, but not a bare stall", async () => { + const withTool = "m-assistant-stalled" + const bare = "m-assistant-bare" + const stalled = new SessionV1.APIError({ + message: + "The model stopped responding: no response headers within 300s (not retried: the attempt had already acted: echo)", + isRetryable: true, + }).toObject() as SessionV1.Assistant["error"] + + const input: SessionV1.WithParts[] = [ + { + info: assistantInfo(withTool, "m-parent", stalled), + parts: [ + { + ...basePart(withTool, "a1"), + type: "tool", + callID: "call-1", + tool: "echo", + state: { + status: "completed", + input: { cmd: "rm" }, + output: "ran rm", + title: "echo", + metadata: {}, + time: { start: 0, end: 1 }, + }, + }, + ] as SessionV1.Part[], + }, + { + info: assistantInfo(bare, "m-parent", stalled), + parts: [{ ...basePart(bare, "b1"), type: "step-start" }] as SessionV1.Part[], + }, + ] + + const out = await MessageV2.toModelMessages(input as unknown as MessageV2.WithParts[], model) + expect(JSON.stringify(out)).toContain("ran rm") + expect(out.filter((m) => m.role === "assistant")).toHaveLength(1) + }) + test("preserves OpenRouter reasoning details through provider transform", async () => { const assistantID = "m-assistant" const openrouterModel: Provider.Model = { diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 20b5c7ccf5..e3992c474b 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -77,6 +77,8 @@ type FakeServer = { requests: number /** Sockets the server saw close (client abort, timeout or our own reset). */ closed: number + /** Indexes of requests whose own socket was seen closing. */ + closedRequests: Set bodies: any[] script: Step[] stop: () => void @@ -85,7 +87,15 @@ type FakeServer = { // A raw TCP/HTTP-1.1 server: Bun.serve cannot produce a real mid-body connection reset, and the // stall shapes below need byte-level control (no headers at all, silent after headers, RST). function fakeServer(script: Step[]): FakeServer { - const state: FakeServer = { url: "", requests: 0, closed: 0, bodies: [], script, stop: () => {} } + const state: FakeServer = { + url: "", + requests: 0, + closed: 0, + closedRequests: new Set(), + bodies: [], + script, + stop: () => {}, + } const sockets = new Set() const frame = (delta: Record, finish: string | null = null) => { const payload = `data: ${JSON.stringify({ id: "c1", object: "chat.completion.chunk", choices: [{ index: 0, delta, finish_reason: finish }] })}\n\n` @@ -130,9 +140,11 @@ function fakeServer(script: Step[]): FakeServer { const server = net.createServer((sock) => { sockets.add(sock) sock.on("error", () => {}) + let requestIndex = -1 sock.on("close", () => { sockets.delete(sock) state.closed++ + if (requestIndex >= 0) state.closedRequests.add(requestIndex) }) let buf = Buffer.alloc(0) let handled = false @@ -147,6 +159,7 @@ function fakeServer(script: Step[]): FakeServer { handled = true state.bodies.push(JSON.parse(buf.subarray(end + 4, end + 4 + length).toString() || "null")) const index = state.requests++ + requestIndex = index // Past the end of the script: behave like a healthy provider so a runaway retry loop shows up as a count. const step = state.script[index] ?? ({ kind: "stream", events: [{ text: "ok" }], end: "stop" } as Step) void respond(sock, step) @@ -212,6 +225,8 @@ type RunOptions = { tools?: Record /** Abort the user signal this long after the request count first reaches 1. */ cancelAfterMs?: number + /** Start the cancel countdown once the first request's socket has been torn down (the stall was detected). */ + cancelAfterStall?: boolean } const servers: FakeServer[] = [] @@ -270,13 +285,21 @@ function runSession(opts: RunOptions) { abort: controller.signal, }) const started = Date.now() + const timers: Array> = [] if (opts.cancelAfterMs !== undefined) { const poll = setInterval(() => { - if (server.requests > 0) { - clearInterval(poll) - setTimeout(() => controller.abort(), opts.cancelAfterMs) - } + const ready = opts.cancelAfterStall ? server.closedRequests.has(0) : server.requests > 0 + if (!ready) return + clearInterval(poll) + timers.push(setTimeout(() => controller.abort(), opts.cancelAfterMs) as any) }, 5) + timers.push(poll) + } + const clearTimers = () => { + for (const t of timers) { + clearInterval(t) + clearTimeout(t as any) + } } const result = yield* Effect.promise(() => handle.process({ @@ -296,8 +319,15 @@ function runSession(opts: RunOptions) { tools: opts.tools ?? {}, abort: controller.signal, } as LLM.StreamInput), - ) + ).pipe(Effect.ensuring(Effect.sync(clearTimers))) const elapsed = Date.now() - started + // the server sees a client-side close asynchronously: wait (bounded) for it rather than asserting instantly + if (opts.cancelAfterMs !== undefined) { + yield* Effect.promise(async () => { + const deadline = Date.now() + 2000 + while (!server.closedRequests.has(0) && Date.now() < deadline) await Bun.sleep(5) + }) + } const parts = MessageV2.parts(msg.id) return { result, parts, message: handle.message, server, elapsed } }), @@ -350,7 +380,7 @@ it.live("request is accepted but never answered: first-byte timeout aborts it an expect(message.error).toBeUndefined() expect(text(parts)).toBe("hello") // the stalled connection was really closed, not merely abandoned - expect(server.closed).toBeGreaterThanOrEqual(1) + expect(server.closedRequests.has(0)).toBe(true) }), ), ) @@ -374,17 +404,17 @@ it.live("goes quiet mid-stream: idle timeout aborts, partial text is discarded, // nothing from the stalled attempt is left behind: one step, one text part expect(countOf(parts, "text")).toBe(1) expect(countOf(parts, "step-start")).toBe(1) - expect(server.closed).toBeGreaterThanOrEqual(1) + expect(server.closedRequests.has(0)).toBe(true) }), ), ) }) -it.live("binary event-stream (Bedrock Converse) responses get the same idle timeout", () => { +it.live("a Bedrock event-stream content type (any letter case) gets the same idle watchdog as SSE", () => { fastBackoff() return runSession({ script: [ - { kind: "stream", contentType: "application/vnd.amazon.eventstream", events: [{ text: "x" }], end: "silent" }, + { kind: "stream", contentType: "Application/Vnd.Amazon.EventStream", events: [{ text: "x" }], end: "silent" }, ok("done"), ], options: FAST, @@ -521,7 +551,7 @@ it.live("user cancel during the first-byte wait ends the session at once and is expect(elapsed).toBeLessThan(3000) // far below the 10s windows: the abort did it expect(server.requests).toBe(1) expect(message.error?.name).toBe("MessageAbortedError") - expect(server.closed).toBeGreaterThanOrEqual(1) + expect(server.closedRequests.has(0)).toBe(true) }), ), ) @@ -532,7 +562,8 @@ it.live("user cancel during the retry backoff ends the session at once and sends return runSession({ script: [{ kind: "silent" }, ok("never reached")], options: FAST, - cancelAfterMs: 800, // first-byte timeout fires at ~400ms; the session is then sleeping in backoff + cancelAfterMs: 50, + cancelAfterStall: true, // the header timeout has fired and the session is sleeping in backoff }).pipe( Effect.tap(({ message, server, elapsed }) => Effect.sync(() => { @@ -627,7 +658,7 @@ it.live("with no headerTimeout configured a 300s first-byte timer is armed and c }) // Direct tests of the discard rule on a hand-built message (the SDK cannot be made to emit these orderings). -const seed = (parts: Array>) => +const seed = (parts: Array>, prior: Array> = []) => provideTmpdirInstanceLegacy((dir) => Effect.gen(function* () { const session = yield* Session.Service @@ -648,6 +679,8 @@ const seed = (parts: Array>) => time: { created: Date.now() }, finish: "end_turn", } as any) + for (const part of prior) + yield* session.updatePart({ id: PartID.ascending(), messageID, sessionID: chat.id, ...part } as any) const before = StallRecovery.partIDs(messageID) for (const part of parts) yield* session.updatePart({ id: PartID.ascending(), messageID, sessionID: chat.id, ...part } as any) @@ -681,6 +714,28 @@ it.live("discard removes streamed text, reasoning, step-start and pending tool i ), ) +it.live("discard leaves parts that predate the failed attempt untouched", () => + seed( + [{ type: "text", text: "from the attempt" }], + [ + { type: "text", text: "earlier step" }, + { + type: "tool", + tool: "echo", + callID: "old", + state: { status: "completed", input: {}, output: "o", title: "t", metadata: {}, time: { start: 1, end: 2 } }, + }, + ], + ).pipe( + Effect.tap(({ result, left }) => + Effect.sync(() => { + expect(result.ok).toBe(true) + expect(left).toEqual(["text", "tool"]) // only the attempt's own text was removed + }), + ), + ), +) + it.live("discard refuses, and removes nothing, once a tool call has started", () => seed([{ type: "text", text: "t" }, pendingTool, runningTool]).pipe( Effect.tap(({ result, left }) => diff --git a/packages/tui/src/feature-plugins/system/notifications.ts b/packages/tui/src/feature-plugins/system/notifications.ts index fbadacce92..eafa5a8c72 100644 --- a/packages/tui/src/feature-plugins/system/notifications.ts +++ b/packages/tui/src/feature-plugins/system/notifications.ts @@ -20,9 +20,13 @@ function notify(api: TuiPluginApi, sessionID: string | undefined, message: strin function sessionErrorMessage(error: SessionError) { if (error?.name === "MessageAbortedError") return "Session aborted" const data = error?.data - if (data && typeof data === "object" && "message" in data && data.message === "SSE read timed out") { - return "Model stopped responding" + // altimate_change start — also recognise the retried-out "The model stopped responding ..." error + if (data && typeof data === "object" && "message" in data && typeof data.message === "string") { + if (data.message === "SSE read timed out" || data.message.startsWith("The model stopped responding")) { + return "Model stopped responding" + } } + // altimate_change end return "Session error" } diff --git a/packages/tui/test/cli/cmd/tui/notifications.test.ts b/packages/tui/test/cli/cmd/tui/notifications.test.ts index 618a225df2..d67826167f 100644 --- a/packages/tui/test/cli/cmd/tui/notifications.test.ts +++ b/packages/tui/test/cli/cmd/tui/notifications.test.ts @@ -22,6 +22,7 @@ async function setup() { subagent: session("subagent", "Subagent session", "session"), abort: session("abort", "Abort session"), timeout: session("timeout", "Timeout session"), + exhausted: session("exhausted", "Exhausted session"), } await Notifications.tui( @@ -225,6 +226,36 @@ describe("internal notifications TUI plugin", () => { ]) }) + test("recognizes the retried-out 'model stopped responding' error", async () => { + const harness = await setup() + + harness.emit({ + id: "event-1", + type: "session.status", + properties: { sessionID: "exhausted", status: { type: "busy" } }, + }) + harness.emit({ + id: "event-2", + type: "session.error", + properties: { + sessionID: "exhausted", + error: { + name: "APIError", + data: { message: "The model stopped responding: no response headers within 300s (gave up after 5 retries)" }, + }, + }, + }) + + expect(harness.notifications).toEqual([ + { + title: "Exhausted session", + message: "Model stopped responding", + notification: { when: "blurred" }, + sound: { name: "error", when: "always" }, + }, + ]) + }) + test("special-cases aborts and model response timeouts", async () => { const harness = await setup() From fc5f18953dbb4cf276f127bbf01d4afd1ab244a2 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 12:59:13 -0700 Subject: [PATCH 07/11] test(tui): type the retried-out stall notification fixture Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/tui/test/cli/cmd/tui/notifications.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/tui/test/cli/cmd/tui/notifications.test.ts b/packages/tui/test/cli/cmd/tui/notifications.test.ts index d67826167f..f558ffb243 100644 --- a/packages/tui/test/cli/cmd/tui/notifications.test.ts +++ b/packages/tui/test/cli/cmd/tui/notifications.test.ts @@ -241,7 +241,10 @@ describe("internal notifications TUI plugin", () => { sessionID: "exhausted", error: { name: "APIError", - data: { message: "The model stopped responding: no response headers within 300s (gave up after 5 retries)" }, + data: { + message: "The model stopped responding: no response headers within 300s (gave up after 5 retries)", + isRetryable: true, + }, }, }, }) From 2613ba519ba6192a0009332ed409ffb304b0916d Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 13:26:19 -0700 Subject: [PATCH 08/11] fix(session): replay only completed tool results of a stalled step Narrow the replay exception for a stalled step to messages where a tool actually completed, so partial text or tool input that never ran is not fed to the next turn. Also wait for the torn-down requests' close events in the stall tests before asserting on them. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/session/message-v2.ts | 14 ++++---- .../opencode/test/session/message-v2.test.ts | 32 ++++++++++++++++--- .../test/session/stall-recovery.test.ts | 21 ++++++++---- 3 files changed, 49 insertions(+), 18 deletions(-) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 51a100adb8..e9e4ec06f0 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -763,15 +763,17 @@ export namespace MessageV2 { const differentModel = `${model.providerID}/${model.id}` !== `${msg.info.providerID}/${msg.info.modelID}` const media: Array<{ mime: string; url: string }> = [] - // altimate_change start — a stalled step whose retry was refused because a tool had already run keeps its - // tool history in replay (as an aborted step does), so the next turn knows those side effects happened + // altimate_change start — a stalled step whose retry was refused keeps the results of tools that actually + // completed in replay, so the next turn knows those side effects happened. Partial text or tool input that + // never ran is not replayed. if ( msg.info.error && !( - (MessageV2.AbortedError.isInstance(msg.info.error) || - (MessageV2.APIError.isInstance(msg.info.error) && - msg.info.error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING))) && - msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning") + (MessageV2.AbortedError.isInstance(msg.info.error) && + msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning")) || + (MessageV2.APIError.isInstance(msg.info.error) && + msg.info.error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) && + msg.parts.some((part) => part.type === "tool" && part.state.status === "completed")) ) ) { continue diff --git a/packages/opencode/test/session/message-v2.test.ts b/packages/opencode/test/session/message-v2.test.ts index ef28db241c..65f617fa36 100644 --- a/packages/opencode/test/session/message-v2.test.ts +++ b/packages/opencode/test/session/message-v2.test.ts @@ -1054,9 +1054,10 @@ describe("session.message-v2.toModelMessage", () => { ]) }) - test("keeps the tool history of a stalled step whose retry was refused, but not a bare stall", async () => { + test("replays a stalled step only when a tool actually completed in it", async () => { const withTool = "m-assistant-stalled" - const bare = "m-assistant-bare" + const textOnly = "m-assistant-text" + const pendingTool = "m-assistant-pending" const stalled = new SessionV1.APIError({ message: "The model stopped responding: no response headers within 300s (not retried: the attempt had already acted: echo)", @@ -1084,13 +1085,34 @@ describe("session.message-v2.toModelMessage", () => { ] as SessionV1.Part[], }, { - info: assistantInfo(bare, "m-parent", stalled), - parts: [{ ...basePart(bare, "b1"), type: "step-start" }] as SessionV1.Part[], + info: assistantInfo(textOnly, "m-parent", stalled), + parts: [{ ...basePart(textOnly, "b1"), type: "text", text: "half an answ" }] as SessionV1.Part[], + }, + { + info: assistantInfo(pendingTool, "m-parent", stalled), + parts: [ + { + ...basePart(pendingTool, "c1"), + type: "tool", + callID: "call-2", + tool: "echo", + state: { + status: "error", + input: {}, + error: "Tool execution aborted", + metadata: { interrupted: true }, + time: { start: 0, end: 1 }, + }, + }, + ] as SessionV1.Part[], }, ] const out = await MessageV2.toModelMessages(input as unknown as MessageV2.WithParts[], model) - expect(JSON.stringify(out)).toContain("ran rm") + const text = JSON.stringify(out) + expect(text).toContain("ran rm") + expect(text).not.toContain("half an answ") + expect(text).not.toContain("Tool execution aborted") expect(out.filter((m) => m.role === "assistant")).toHaveLength(1) }) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index e3992c474b..72e728d5cb 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -321,13 +321,20 @@ function runSession(opts: RunOptions) { } as LLM.StreamInput), ).pipe(Effect.ensuring(Effect.sync(clearTimers))) const elapsed = Date.now() - started - // the server sees a client-side close asynchronously: wait (bounded) for it rather than asserting instantly - if (opts.cancelAfterMs !== undefined) { - yield* Effect.promise(async () => { - const deadline = Date.now() + 2000 - while (!server.closedRequests.has(0) && Date.now() < deadline) await Bun.sleep(5) - }) - } + // The server sees a client-side close asynchronously. Every request but the last was torn down by the + // client (timeout, reset or cancel): wait, bounded, for those close events before anyone asserts on them. + yield* Effect.promise(async () => { + const deadline = Date.now() + 2000 + const settled = () => + Array.from({ length: Math.max(0, server.requests - 1) }, (_, i) => i).every((i) => + server.closedRequests.has(i), + ) + while ( + (!settled() || (opts.cancelAfterMs !== undefined && !server.closedRequests.has(0))) && + Date.now() < deadline + ) + await Bun.sleep(5) + }) const parts = MessageV2.parts(msg.id) return { result, parts, message: handle.message, server, elapsed } }), From f4a5cc725158489d594a81e6ddfbbaba9b5d8db9 Mon Sep 17 00:00:00 2001 From: Anand Gupta Date: Wed, 7 Oct 2026 13:40:04 -0700 Subject: [PATCH 09/11] revert(session): drop the stalled-step replay exception The replay of an errored assistant step is a pre-existing limitation for every error, and handling it correctly (per-part filtering, all retryable errors, errored tools) is a separate change. Keep this PR to stall detection and the retry rules; the known limit is documented in the PR. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/session/message-v2.ts | 11 +--- .../opencode/test/session/message-v2.test.ts | 62 ------------------- 2 files changed, 2 insertions(+), 71 deletions(-) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index e9e4ec06f0..2f2d54043f 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -763,22 +763,15 @@ export namespace MessageV2 { const differentModel = `${model.providerID}/${model.id}` !== `${msg.info.providerID}/${msg.info.modelID}` const media: Array<{ mime: string; url: string }> = [] - // altimate_change start — a stalled step whose retry was refused keeps the results of tools that actually - // completed in replay, so the next turn knows those side effects happened. Partial text or tool input that - // never ran is not replayed. if ( msg.info.error && !( - (MessageV2.AbortedError.isInstance(msg.info.error) && - msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning")) || - (MessageV2.APIError.isInstance(msg.info.error) && - msg.info.error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING) && - msg.parts.some((part) => part.type === "tool" && part.state.status === "completed")) + MessageV2.AbortedError.isInstance(msg.info.error) && + msg.parts.some((part) => part.type !== "step-start" && part.type !== "reasoning") ) ) { continue } - // altimate_change end const assistantMessage: UIMessage = { id: msg.info.id, role: "assistant", diff --git a/packages/opencode/test/session/message-v2.test.ts b/packages/opencode/test/session/message-v2.test.ts index 65f617fa36..ca87e6fb29 100644 --- a/packages/opencode/test/session/message-v2.test.ts +++ b/packages/opencode/test/session/message-v2.test.ts @@ -1054,68 +1054,6 @@ describe("session.message-v2.toModelMessage", () => { ]) }) - test("replays a stalled step only when a tool actually completed in it", async () => { - const withTool = "m-assistant-stalled" - const textOnly = "m-assistant-text" - const pendingTool = "m-assistant-pending" - const stalled = new SessionV1.APIError({ - message: - "The model stopped responding: no response headers within 300s (not retried: the attempt had already acted: echo)", - isRetryable: true, - }).toObject() as SessionV1.Assistant["error"] - - const input: SessionV1.WithParts[] = [ - { - info: assistantInfo(withTool, "m-parent", stalled), - parts: [ - { - ...basePart(withTool, "a1"), - type: "tool", - callID: "call-1", - tool: "echo", - state: { - status: "completed", - input: { cmd: "rm" }, - output: "ran rm", - title: "echo", - metadata: {}, - time: { start: 0, end: 1 }, - }, - }, - ] as SessionV1.Part[], - }, - { - info: assistantInfo(textOnly, "m-parent", stalled), - parts: [{ ...basePart(textOnly, "b1"), type: "text", text: "half an answ" }] as SessionV1.Part[], - }, - { - info: assistantInfo(pendingTool, "m-parent", stalled), - parts: [ - { - ...basePart(pendingTool, "c1"), - type: "tool", - callID: "call-2", - tool: "echo", - state: { - status: "error", - input: {}, - error: "Tool execution aborted", - metadata: { interrupted: true }, - time: { start: 0, end: 1 }, - }, - }, - ] as SessionV1.Part[], - }, - ] - - const out = await MessageV2.toModelMessages(input as unknown as MessageV2.WithParts[], model) - const text = JSON.stringify(out) - expect(text).toContain("ran rm") - expect(text).not.toContain("half an answ") - expect(text).not.toContain("Tool execution aborted") - expect(out.filter((m) => m.role === "assistant")).toHaveLength(1) - }) - test("preserves OpenRouter reasoning details through provider transform", async () => { const assistantID = "m-assistant" const openrouterModel: Provider.Model = { From b852ba47d68bb71819a16fa166c8b563d9cf0017 Mon Sep 17 00:00:00 2001 From: Anand Gupta Date: Wed, 7 Oct 2026 14:12:08 -0700 Subject: [PATCH 10/11] fix(session): track tool execution start, bound custom-fetch hangs, clean up atomically - The retry guard now also refuses once any tool's execute() began during the attempt, counted in the processor from the tool wrapper's `beginToolExecution`, because the AI SDK can start a tool before its part leaves `pending` or its tool-call event is persisted. - Remove a failed attempt's parts in one delete statement so a cleanup failure cannot leave the attempt half deleted. - Race the whole custom fetch against the first-byte deadline, so work such as Vertex credential acquisition that ignores the abort signal is bounded. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/provider/provider.ts | 25 ++++++++++++- packages/opencode/src/session/processor.ts | 7 ++++ .../opencode/src/session/stall-recovery.ts | 37 ++++++++++++++++--- .../chunk-timeout-844.test.ts | 24 ++++++++++++ .../test/session/stall-recovery.test.ts | 36 ++++++++++++++++++ 5 files changed, 121 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 88fc1f0eac..95c200eb4d 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -2028,12 +2028,33 @@ 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 (headerTimeoutCtl) { + const signal = headerTimeoutCtl.signal + let onAbort = () => {} + const deadline = new Promise((_, reject) => { + onAbort = () => reject(signal.reason) + signal.addEventListener("abort", onAbort, { once: true }) + }) + pending.catch(() => {}) // the loser of the race must not surface as an unhandled rejection + try { + res = await Promise.race([pending, deadline]) + } finally { + signal.removeEventListener("abort", onAbort) + headerTimeoutCtl.clear() + } + } else { + res = await pending + } + // altimate_change end if (!chunkAbortCtl) return res return wrapSSE(res, chunkTimeout, chunkAbortCtl) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 5658e2d930..2e0ce9e64a 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -303,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 @@ -320,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) { @@ -430,6 +435,7 @@ export namespace SessionProcessor { while (true) { // altimate_change start — remember which parts predate this attempt so a failed one can be unwound let partsBeforeAttempt = new Set() + const executionsBeforeAttempt = toolExecutionsStarted // altimate_change end try { // altimate_change start — see above @@ -1231,6 +1237,7 @@ export namespace SessionProcessor { 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 }) diff --git a/packages/opencode/src/session/stall-recovery.ts b/packages/opencode/src/session/stall-recovery.ts index a2577ac2c6..2300799998 100644 --- a/packages/opencode/src/session/stall-recovery.ts +++ b/packages/opencode/src/session/stall-recovery.ts @@ -1,6 +1,8 @@ import { Log } from "@/util/log" import { MessageV2 } from "./message-v2" -import { Session } from "." +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" /** @@ -31,7 +33,10 @@ export async function discardAttempt(input: { sessionID: SessionID messageID: MessageID before: ReadonlySet + /** A tool's execute() began during the attempt (tracked by the tool wrapper, not inferred from part status). */ + toolExecutionStarted?: boolean }): Promise { + 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"), @@ -39,11 +44,31 @@ export async function discardAttempt(input: { if (acted.length > 0) { return { ok: false, dispatched: acted.map((part) => (part.type === "tool" ? part.tool : part.type)) } } - const removed = new Set() - for (const part of added) { - if (!DISCARDED_WHEN_RETRIED.has(part.type)) continue - await Session.removePart({ sessionID: input.sessionID, messageID: input.messageID, partID: part.id }) - removed.add(part.id) + // 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(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 }) diff --git a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts index 239024e318..88194de004 100644 --- a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts +++ b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts @@ -272,3 +272,27 @@ describe("PR #844 — DEFAULT_CHUNK_TIMEOUT 2min→5min", () => { // Capture the real setTimeout up front so gap-5's wait is unaffected by any per-test stubbing. const origFetchTimeout = globalThis.setTimeout + +describe("first-byte timeout bounds fetch work that ignores the abort signal", () => { + beforeAll(() => prepareReleaseValidationDatabase()) + + // Google Vertex's custom fetch awaits credential acquisition before it ever touches the signal. + test("a fetch that never settles and never observes the signal still rejects at the deadline", async () => { + await using tmp = await tmpdir({ init: (dir) => writeCapturingProvider(dir, { apiKey: "x", headerTimeout: 150 }) }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const wrapper = await captureWrapper() + const origFetch = globalThis.fetch + globalThis.fetch = (() => new Promise(() => {})) as any + try { + const started = Date.now() + await expect(wrapper("http://x/v1/chat", { method: "POST" })).rejects.toThrow("headers timed out") + expect(Date.now() - started).toBeLessThan(3000) + } finally { + globalThis.fetch = origFetch + } + }, + }) + }) +}) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 72e728d5cb..44cca9caad 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -227,6 +227,8 @@ type RunOptions = { cancelAfterMs?: number /** Start the cancel countdown once the first request's socket has been torn down (the stall was detected). */ cancelAfterStall?: boolean + /** Called once, as soon as the server has the first request (e.g. to simulate the SDK starting a tool). */ + onFirstRequest?: (handle: SessionProcessor.Info) => void } const servers: FakeServer[] = [] @@ -295,6 +297,14 @@ function runSession(opts: RunOptions) { }, 5) timers.push(poll) } + if (opts.onFirstRequest) { + const poll = setInterval(() => { + if (server.requests === 0) return + clearInterval(poll) + opts.onFirstRequest!(handle) + }, 5) + timers.push(poll) + } const clearTimers = () => { for (const t of timers) { clearInterval(t) @@ -530,6 +540,32 @@ it.live("a tool call already dispatched is never re-run: the stall is reported i ) }) +it.live("a tool execution that began while its part was still pending is never re-run: no retry", () => { + fastBackoff() + const calls: string[] = [] + return runSession({ + tools: echoTool(calls), + script: [ + // only a partial tool input reaches the processor: its part stays pending and no tool-call event exists + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"ec' } }], end: "silent" }, + { kind: "stream", events: [{ tool: { id: "call_1", name: "echo", args: '{"cmd":"go"}' } }], end: "tool_calls" }, + ], + options: FAST, + // what prompt.ts's tool wrapper does when the SDK starts execute(), ahead of any persisted state change + onFirstRequest: (handle) => void handle.beginToolExecution("call_1"), + }).pipe( + Effect.tap(({ message, server, parts }) => + Effect.sync(() => { + expect(server.requests).toBe(1) + expect(calls).toEqual([]) + expect(message.error?.name).toBe("APIError") + expect((message.error?.data as { message: string }).message).toContain("not retried") + expect(parts.some((p) => p.type === "tool")).toBe(true) // nothing was deleted + }), + ), + ) +}) + it.live("retries are bounded and exhaustion ends the session with a clear 'model stopped responding' error", () => { fastBackoff() return runSession({ script: Array.from({ length: 20 }, () => ({ kind: "silent" }) as Step), options: FAST }).pipe( From a689e5ad8cc518906cc1e27be7ab8367005840e0 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Wed, 7 Oct 2026 14:25:50 -0700 Subject: [PATCH 11/11] fix(provider): settle a stuck custom fetch on caller cancel too Race the custom fetch against the combined request signal (first-byte deadline, caller cancel, timeouts), not only the deadline, so a Stop is not held up by pre-request work that ignores the signal. The tool-race test now waits for the pending tool part before reporting the execution started. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/opencode/src/provider/provider.ts | 21 ++++++++++------- .../chunk-timeout-844.test.ts | 23 +++++++++++++++++++ .../test/session/stall-recovery.test.ts | 4 +++- 3 files changed, 39 insertions(+), 9 deletions(-) diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 95c200eb4d..908c6fed2c 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -2037,22 +2037,27 @@ export namespace Provider { timeout: false, }))() let res: Response - if (headerTimeoutCtl) { - const signal = headerTimeoutCtl.signal - let onAbort = () => {} + 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((_, reject) => { - onAbort = () => reject(signal.reason) - signal.addEventListener("abort", onAbort, { once: true }) + 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 { - signal.removeEventListener("abort", onAbort) - headerTimeoutCtl.clear() + for (const cleanup of cleanups) cleanup() + headerTimeoutCtl?.clear() } } else { - res = await pending + res = await pending.finally(() => headerTimeoutCtl?.clear()) } // altimate_change end diff --git a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts index 88194de004..39492c5103 100644 --- a/packages/opencode/test/release-validation/chunk-timeout-844.test.ts +++ b/packages/opencode/test/release-validation/chunk-timeout-844.test.ts @@ -295,4 +295,27 @@ describe("first-byte timeout bounds fetch work that ignores the abort signal", ( }, }) }) + test("a caller cancel also settles a fetch that ignores the signal", async () => { + await using tmp = await tmpdir({ + init: (dir) => writeCapturingProvider(dir, { apiKey: "x", headerTimeout: 60_000 }), + }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const wrapper = await captureWrapper() + const origFetch = globalThis.fetch + globalThis.fetch = (() => new Promise(() => {})) as any + try { + const ctl = new AbortController() + const started = Date.now() + const call = wrapper("http://x/v1/chat", { method: "POST", signal: ctl.signal }) + setTimeout(() => ctl.abort(new DOMException("Aborted", "AbortError")), 100) + await expect(call).rejects.toThrow("Aborted") + expect(Date.now() - started).toBeLessThan(3000) // far below the 60s deadline + } finally { + globalThis.fetch = origFetch + } + }, + }) + }) }) diff --git a/packages/opencode/test/session/stall-recovery.test.ts b/packages/opencode/test/session/stall-recovery.test.ts index 44cca9caad..1ff1779068 100644 --- a/packages/opencode/test/session/stall-recovery.test.ts +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -299,7 +299,9 @@ function runSession(opts: RunOptions) { } if (opts.onFirstRequest) { const poll = setInterval(() => { - if (server.requests === 0) return + // only once the stalled attempt has persisted its (still pending) tool part + const pending = MessageV2.parts(msg.id).some((p) => p.type === "tool" && p.state.status === "pending") + if (server.requests === 0 || !pending) return clearInterval(poll) opts.onFirstRequest!(handle) }, 5)