diff --git a/docs/docs/configure/providers.md b/docs/docs/configure/providers.md index 38ee4c2dd5..10e83d44f6 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` turns the check off, and a server that then never answers hangs the session | +| `options.chunkTimeout` | `number` | Milliseconds of silence on a streaming response before the request is aborted and retried (default `300000`). Applies to SSE and Amazon Bedrock event streams | + +### Stalled requests + +If the provider accepts a request and then stops answering, Altimate Code aborts the connection when +`headerTimeout` or `chunkTimeout` elapses and retries the request with the usual backoff, up to five +times. Whatever the stalled attempt had already streamed (text, reasoning, a tool call still being +written) is discarded so the retry does not duplicate it. A tool call that had already been dispatched +is never re-run: the session stops with "The model stopped responding" and names the tool instead of +retrying. When retries run out the session ends with the same message. + +```json +{ + "provider": { + "amazon-bedrock": { "options": { "headerTimeout": 120000, "chunkTimeout": 120000 } } + } +} +``` 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..908c6fed2c 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") ?? "").toLowerCase() + if (!contentType.includes("text/event-stream") && !contentType.includes("application/vnd.amazon.eventstream")) + return res + // altimate_change end const reader = res.body.getReader() const body = new ReadableStream({ @@ -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"] @@ -2014,12 +2028,38 @@ export namespace Provider { } } - const res = await (async () => + // altimate_change start — the deadline must also bound a custom fetch's own pre-request work (for example + // Vertex's credential acquisition), which never observes the signal: race it against the deadline. + const pending = (async () => fetchFn(input, { ...opts, // @ts-ignore see here: https://github.com/oven-sh/bun/issues/16682 timeout: false, - }))().finally(() => headerTimeoutCtl?.clear()) + }))() + let res: Response + if (opts.signal) { + // settle on the combined abort (first-byte deadline, caller cancel, timeouts), whichever comes first + const guards = [opts.signal] + const cleanups: Array<() => void> = [] + const deadline = new Promise((_, reject) => { + for (const guard of guards) { + const onAbort = () => reject(guard.reason) + if (guard.aborted) return onAbort() + guard.addEventListener("abort", onAbort, { once: true }) + cleanups.push(() => guard.removeEventListener("abort", onAbort)) + } + }) + pending.catch(() => {}) // the loser of the race must not surface as an unhandled rejection + try { + res = await Promise.race([pending, deadline]) + } finally { + for (const cleanup of cleanups) cleanup() + headerTimeoutCtl?.clear() + } + } else { + res = await pending.finally(() => headerTimeoutCtl?.clear()) + } + // altimate_change end if (!chunkAbortCtl) return res return wrapSSE(res, chunkTimeout, chunkAbortCtl) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 152b75774b..2f2d54043f 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -1171,6 +1171,30 @@ export namespace MessageV2 { }, { cause: e }, ).toObject() + // altimate_change start — a request the provider stopped answering (no response headers, or a silent + // stream) is a transient stall: classify it as a retryable APIError that names the problem instead of + // surfacing it as an opaque UnknownError that ends the session. Mirrors upstream OpenCode + // (dev: message-v2.ts HeaderTimeoutError / ResponseStreamError cases), with clearer wording and + // limited to the idle watchdog: other ResponseStreamErrors (websocket transport) keep their old handling. + case e instanceof ProviderError.HeaderTimeoutError: + return new MessageV2.APIError( + { + message: `${ProviderError.MODEL_STOPPED_RESPONDING}: no response headers within ${Math.round(e.ms / 1000)}s`, + isRetryable: true, + metadata: { code: e.name, timeoutMs: String(e.ms) }, + }, + { cause: e }, + ).toObject() + case e instanceof ProviderError.ResponseStreamError && e.message === ProviderError.SSE_IDLE_MESSAGE: + return new MessageV2.APIError( + { + message: `${ProviderError.MODEL_STOPPED_RESPONDING}: the response stream went silent (${e.message})`, + isRetryable: true, + metadata: { code: e.name }, + }, + { cause: e }, + ).toObject() + // altimate_change end case APICallError.isInstance(e): const parsed = ProviderError.parseAPICallError({ providerID: ctx.providerID, diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 81cfd3036f..2e0ce9e64a 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" @@ -233,6 +237,22 @@ export namespace SessionProcessor { executionID(execution: ToolExecution) { return allocated.get(keyOf(execution.raw))?.[execution.occurrence] }, + // Forget tool inputs that started streaming but were never called (a discarded attempt) as if they never + // happened: drop their allocations so the per-id slot ordinals that settled()/executionID() index by stay + // aligned, and free their names. Ids allocated by earlier calls stay reserved, so none is ever reused. + discardUnstarted() { + for (const [key, ids] of started) { + const list = allocated.get(key) ?? [] + for (const id of ids) { + const index = list.lastIndexOf(id) + if (index >= 0) list.splice(index, 1) + used.delete(id) + occurrences.set(key, Math.max(0, (occurrences.get(key) ?? 1) - 1)) + } + if (list.length === 0) allocated.delete(key) + } + started.clear() + }, }) } // altimate_change end @@ -283,6 +303,7 @@ export namespace SessionProcessor { // session thinking it's stuck. Track whether the session has ever produced a // tool call; if plan agent finishes its first step with stop-no-tools, warn. let sessionToolCallsMade = 0 + let toolExecutionsStarted = 0 let planNoToolWarningEmitted = false // altimate_change end @@ -300,6 +321,10 @@ export namespace SessionProcessor { return toolcalls.get(matched ?? coerceToolCallID.peek(toolCallID)) }, beginToolExecution(toolCallID: string) { + // altimate_change start — the SDK runs a tool as soon as its input is complete, possibly before its part + // leaves `pending` (or is persisted at all): this counter is the source of truth for "a tool has started" + toolExecutionsStarted++ + // altimate_change end return coerceToolCallID.beginExecution(toolCallID) }, finishToolExecution(execution: ToolExecution) { @@ -408,7 +433,14 @@ export namespace SessionProcessor { } // altimate_change end while (true) { + // altimate_change start — remember which parts predate this attempt so a failed one can be unwound + let partsBeforeAttempt = new Set() + const executionsBeforeAttempt = toolExecutionsStarted + // altimate_change end try { + // altimate_change start — see above + partsBeforeAttempt = StallRecovery.partIDs(input.assistantMessage.id) + // altimate_change end let currentText: MessageV2.TextPart | undefined let reasoningMap: Record = {} if (snapshot === undefined) { @@ -1195,8 +1227,29 @@ export namespace SessionProcessor { } } else { const retry = SessionRetry.retryable(error) - // altimate_change start — cap retries to avoid infinite loops, log on exhaustion - if (retry !== undefined && attempt < SessionRetry.RETRY_MAX_ATTEMPTS) { + // altimate_change start — cap retries to avoid infinite loops, log on exhaustion. + // A user cancel is never retried; and a retry only goes ahead after the failed attempt's + // partial output is discarded, which is refused once a tool call was dispatched (see + // StallRecovery) so a retry can neither duplicate streamed output nor re-run a tool. + const discard = + retry !== undefined && attempt < SessionRetry.RETRY_MAX_ATTEMPTS && !input.abort.aborted + ? await StallRecovery.discardAttempt({ + sessionID: input.sessionID, + messageID: input.assistantMessage.id, + before: partsBeforeAttempt, + toolExecutionStarted: toolExecutionsStarted > executionsBeforeAttempt, + }).catch((err) => { + // a failed cleanup must not escape the catch block: fall through to the terminal path + log.warn("could not discard partial output; not retrying", { error: err }) + return undefined + }) + : undefined + if (discard && !discard.ok) { + log.warn("not retrying: the attempt already acted", { parts: discard.dispatched }) + } + if (retry !== undefined && discard?.ok) { + for (const [id, part] of toolcalls) if (discard.removed.has(part.id)) toolcalls.delete(id) + coerceToolCallID.discardUnstarted() // altimate_change end attempt++ const delay = SessionRetry.delay(attempt, error.name === "APIError" ? error : undefined) @@ -1212,7 +1265,8 @@ export namespace SessionProcessor { continue } // altimate_change start — log when retries exhausted for debugging - if (retry !== undefined) { + const exhausted = retry !== undefined && attempt >= SessionRetry.RETRY_MAX_ATTEMPTS + if (exhausted) { log.warn("max retry attempts reached, giving up", { attempt, message: retry, @@ -1221,6 +1275,15 @@ export namespace SessionProcessor { }) } // altimate_change end + // altimate_change start — say why a retryable error was not retried or retried out + if (MessageV2.APIError.isInstance(error)) { + if (discard && !discard.ok) { + error.data.message += ` (not retried: the attempt had already acted: ${discard.dispatched.join(", ")})` + } else if (exhausted && error.data.message.startsWith(ProviderError.MODEL_STOPPED_RESPONDING)) { + error.data.message += ` (gave up after ${attempt} retries)` + } + } + // altimate_change end input.assistantMessage.error = error // altimate_change start — order terminal error before idle publication // Publish the error before idle. The run harness drains this 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 new file mode 100644 index 0000000000..2300799998 --- /dev/null +++ b/packages/opencode/src/session/stall-recovery.ts @@ -0,0 +1,78 @@ +import { Log } from "@/util/log" +import { MessageV2 } from "./message-v2" +import { Bus } from "@/bus" +import { Database, and, eq, inArray } from "@/storage/db" +import { PartTable } from "@opencode-ai/core/session/sql" +import type { MessageID, SessionID } from "./schema" + +/** + * Cleanup that makes it safe to re-issue a model request inside the same assistant message after the + * previous attempt failed part-way (stalled, reset, or errored mid-stream). + * + * Rule: a retry may only replace what the failed attempt streamed, never what it already acted on. + * - text, reasoning and step-start parts written by the failed attempt are removed, and so are tool parts + * still `pending` (input was streaming, the tool never started). The retried request regenerates them. + * - if the failed attempt left any tool part that STARTED (running, completed or errored), the tool call was + * dispatched and may have had side effects. Re-requesting would make the model emit it again, so the attempt + * is NOT retried and nothing is removed: the error surfaces and the dispatched call keeps its record. + * - the same holds if the attempt already finished its step (a step-finish part exists): its cost and tokens + * are on the message, so a retry would count them twice and answer a second time. + */ +const log = Log.create({ service: "session.stall-recovery" }) + +// A `tool` part reaching this set is necessarily still `pending` (started ones are rejected first). +const DISCARDED_WHEN_RETRIED: ReadonlySet = 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 + /** 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"), + ) + if (acted.length > 0) { + return { ok: false, dispatched: acted.map((part) => (part.type === "tool" ? part.tool : part.type)) } + } + // One statement, so a failure leaves the attempt's output intact rather than half deleted. + const stale = added.filter((part) => DISCARDED_WHEN_RETRIED.has(part.type)) + const removed = new Set(stale.map((part) => part.id)) + if (stale.length > 0) { + Database.use((db) => { + db.delete(PartTable) + .where( + and( + eq(PartTable.session_id, input.sessionID), + inArray( + PartTable.id, + stale.map((part) => part.id), + ), + ), + ) + .run() + for (const part of stale) + Database.effect(() => + Bus.publish(MessageV2.Event.PartRemoved, { + sessionID: input.sessionID, + messageID: input.messageID, + partID: part.id, + }), + ) + }) + } + if (removed.size > 0) + log.info("discarded partial output before retry", { messageID: input.messageID, parts: removed.size }) + return { ok: true, removed } +} + +export * as StallRecovery from "./stall-recovery" 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..39492c5103 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") ?? "").toLowerCase()') + expect(normalized).toContain( + '!contentType.includes("text/event-stream") && !contentType.includes("application/vnd.amazon.eventstream")', + ) }) }) @@ -268,3 +272,50 @@ 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 + } + }, + }) + }) + 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/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 new file mode 100644 index 0000000000..1ff1779068 --- /dev/null +++ b/packages/opencode/test/session/stall-recovery.test.ts @@ -0,0 +1,875 @@ +// 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 { 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" +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 + /** Indexes of requests whose own socket was seen closing. */ + closedRequests: Set + 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, + 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` + 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", () => {}) + 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 + 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++ + 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) + }) + }) + 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 { + // 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", + 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 + /** 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[] = [] +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() + const timers: Array> = [] + if (opts.cancelAfterMs !== undefined) { + const poll = setInterval(() => { + 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) + } + if (opts.onFirstRequest) { + const poll = setInterval(() => { + // 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) + timers.push(poll) + } + const clearTimers = () => { + for (const t of timers) { + clearInterval(t) + clearTimeout(t as any) + } + } + 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), + ).pipe(Effect.ensuring(Effect.sync(clearTimers))) + const elapsed = Date.now() - started + // 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 } + }), + { 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.closedRequests.has(0)).toBe(true) + }), + ), + ) +}) + +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.closedRequests.has(0)).toBe(true) + }), + ), + ) +}) + +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" }, + 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("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( + 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.closedRequests.has(0)).toBe(true) + }), + ), + ) +}) + +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: 50, + cancelAfterStall: true, // the header timeout has fired and the session is 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 + }), + ), + ) +}) + +// Direct tests of the discard rule on a hand-built message (the SDK cannot be made to emit these orderings). +const seed = (parts: Array>, prior: 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) + 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) + 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 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 }) => + 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("tool-call id bookkeeping across a discarded attempt", () => { + 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") + 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(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) + }) +}) + +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 (websocket transport) keep their previous, non-retried handling", () => { + const error = MessageV2.fromError(new ProviderError.ResponseStreamError("WebSocket closed"), { providerID }) + expect(error.name).toBe("UnknownError") + expect(SessionRetry.retryable(error)).toBeUndefined() + }) + + 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() + }) +}) 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..f558ffb243 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,39 @@ 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)", + isRetryable: true, + }, + }, + }, + }) + + 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()