From 5ff83bf6b3d397561ec83fdf429c8748a3bacdfa Mon Sep 17 00:00:00 2001 From: Sarath S Menon Date: Wed, 7 Oct 2026 13:14:11 +0000 Subject: [PATCH 1/2] fix(session): recover interrupted V4 model streams safely Recognize gateway transport failures and cap retries before tool execution, preserving earlier parts. The focused SDK regression fails on main and passes with recovery, exhaustion, action and cancellation guards. --- AGENTS.md | 2 + .../verification/v4-llm-transport-recovery.md | 15 ++ packages/opencode/src/provider/error.ts | 16 ++ packages/opencode/src/session/llm.ts | 29 ++- packages/opencode/src/session/message-v2.ts | 1 + packages/opencode/src/session/processor.ts | 31 ++- packages/opencode/src/session/retry.ts | 8 +- packages/opencode/test/lib/llm-server.ts | 1 + .../test/session/processor-effect.test.ts | 214 +++++++++++++++++- 9 files changed, 304 insertions(+), 13 deletions(-) create mode 100644 docs/verification/v4-llm-transport-recovery.md diff --git a/AGENTS.md b/AGENTS.md index 617b6093de..a881621935 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -233,3 +233,5 @@ runs this filtered command. - Keep delivery vocabulary explicit. Prompts steer by default and promote at the next safe provider-turn boundary while the current drain requires continuation. An explicit `queue` input remains pending until the Session would otherwise become idle; promote one queued input at that boundary, then reevaluate continuation before promoting another. Promoting any new user input resets the selected agent's provider-turn allowance; a batch of steers resets it once. - Keep EventV2 replay owner claims separate from clustered Session execution ownership. - Keep the System Context algebra, registry, and built-ins in `src/system-context`; keep Context Source producers with their observed domains, and keep Session History selection plus Context Epoch persistence Session-owned. + +- Retry streamed transport failures with a finite budget before tool execution starts; preserve earlier parts and discard only the incomplete attempt. diff --git a/docs/verification/v4-llm-transport-recovery.md b/docs/verification/v4-llm-transport-recovery.md new file mode 100644 index 0000000000..05151398e0 --- /dev/null +++ b/docs/verification/v4-llm-transport-recovery.md @@ -0,0 +1,15 @@ +# V4 LLM transport recovery + +The Cloud proxy emits `upstream_stream_failed` after an interrupted upstream stream. BrowserCode previously treated this envelope as an unknown terminal error. The fix recognizes transport exception classes, permits at most three total attempts with existing backoff, removes only incomplete attempt parts, and refuses replay once prepared tool execution starts or a tool call is observed. + +The focused test uses the actual OpenAI SDK and an HTTP 200 SSE fixture. It covers ReadError, RemoteProtocolError, APIConnectionError, exhaustion, an executed action, a non-transport exception, and interruption during backoff. Recovery requires two HTTP calls, one tool execution, recovered output only, and correct token/cost accounting. Exhaustion requires exactly three HTTP calls. + +With the final unchanged regression (SHA256 `2caa9604b5ce2aaa90b5449e0e74a4c0411e08a9fd81c915955dcff3e1a67db3`), baseline main had two passing safety cases and five failures; the fix passed all seven. Package typecheck passed. Focused lint reported zero errors and 42 warnings. VM evidence is in `/home/exedev/v4-llm-transport-proof`. + +Run from `packages/opencode` with Bun 1.3.14: +```sh +bun test test/session/processor-effect.test.ts -t 'gateway transport recovery' --timeout 25000 +bun typecheck +``` + +This fixes a recovery gap, not the unconfirmed provider/network trigger. Production still uses BrowserCode 0.1.21-telemetry.1; an updated binary release and a reviewed Cloud V4 worker pin are required. No release, Cloud pin, AWS change, or deployment was performed. diff --git a/packages/opencode/src/provider/error.ts b/packages/opencode/src/provider/error.ts index 5214af43df..132315472a 100644 --- a/packages/opencode/src/provider/error.ts +++ b/packages/opencode/src/provider/error.ts @@ -97,6 +97,7 @@ export type ParsedStreamError = message: string isRetryable: boolean responseBody: string + metadata?: Record } export function parseStreamError(input: unknown): ParsedStreamError | undefined { @@ -142,6 +143,21 @@ export function parseStreamError(input: unknown): ParsedStreamError | undefined isRetryable: false, responseBody, } + case "upstream_stream_failed": + if ( + typeof error.message !== "string" || + !/^(ReadError|ReadTimeout|RemoteProtocolError|APIConnectionError|ConnectError|ConnectTimeout)(?::|$)/.test( + error.message, + ) + ) + return + return { + type: "api_error", + message: error.message, + isRetryable: true, + responseBody, + metadata: { code: "upstream_stream_failed" }, + } case "server_is_overloaded": case "server_error": return { diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index 3fe9509015..db1336474d 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -44,6 +44,7 @@ export type StreamInput = { messages: ModelMessage[] small?: boolean tools: Record + onToolExecution?: () => void retries?: number toolChoice?: "auto" | "required" | "none" } @@ -115,6 +116,22 @@ const live: Layer.Layer< isWorkflow, }) + // SDK tool execution can begin before the processor receives its tool-call event. + const tools = Object.fromEntries( + Object.entries(prepared.tools).map(([name, value]) => [ + name, + value.execute + ? { + ...value, + execute: (...args: Parameters>) => { + input.onToolExecution?.() + return value.execute!(...args) + }, + } + : value, + ]), + ) + // Wire up toolExecutor for DWS workflow models so that tool calls // from the workflow service are executed via opencode's tool system // and results sent back over the WebSocket. @@ -128,7 +145,7 @@ const live: Layer.Layer< workflowModel.sessionID = input.sessionID workflowModel.systemPrompt = prepared.system.join("\n") workflowModel.toolExecutor = async (toolName, argsJson, _requestID) => { - const t = prepared.tools[toolName] + const t = tools[toolName] if (!t || !t.execute) { return { result: "", error: `Unknown tool: ${toolName}` } } @@ -151,7 +168,7 @@ const live: Layer.Layer< } const ruleset = Permission.merge(input.agent.permission ?? [], input.permission ?? []) - workflowModel.sessionPreapprovedTools = Object.keys(prepared.tools).filter((name) => { + workflowModel.sessionPreapprovedTools = Object.keys(tools).filter((name) => { const match = ruleset.findLast((rule) => Wildcard.match(name, rule.permission)) return !match || match.action !== "ask" }) @@ -234,7 +251,7 @@ const live: Layer.Layer< auth: info, llmClient, messages: prepared.messages, - tools: prepared.tools, + tools, toolChoice: input.toolChoice, temperature: prepared.params.temperature, topP: prepared.params.topP, @@ -299,7 +316,7 @@ const live: Layer.Layer< includeRawChunks: input.model.providerID.includes("github-copilot"), async experimental_repairToolCall(failed) { const lower = failed.toolCall.toolName.toLowerCase() - if (lower !== failed.toolCall.toolName && prepared.tools[lower]) { + if (lower !== failed.toolCall.toolName && tools[lower]) { return { ...failed.toolCall, toolName: lower, @@ -319,8 +336,8 @@ const live: Layer.Layer< topP: prepared.params.topP, topK: prepared.params.topK, providerOptions: ProviderTransform.providerOptions(input.model, prepared.params.options), - activeTools: Object.keys(prepared.tools).filter((x) => x !== "invalid"), - tools: prepared.tools, + activeTools: Object.keys(tools).filter((x) => x !== "invalid"), + tools, toolChoice: input.toolChoice, maxOutputTokens: prepared.params.maxOutputTokens, abortSignal: input.abort, diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index be8bce8d80..ca959d6b44 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -747,6 +747,7 @@ export function fromError( message: parsed.message, isRetryable: parsed.isRetryable, responseBody: parsed.responseBody, + ...(parsed.metadata ? { metadata: parsed.metadata } : {}), }, { cause: e, diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 15e74557c3..2dda80877d 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -167,14 +167,14 @@ const layer = Layer.effect( return { call, part } }) - const resetOutputLimit = Effect.fn("SessionProcessor.resetOutputLimit")(function* () { + const resetAttempt = Effect.fn("SessionProcessor.resetAttempt")(function* (keep?: ReadonlySet) { const parts = yield* MessageV2.parts(ctx.assistantMessage.id).pipe( Effect.provideService(Database.Service, database), ) // Replace the streamed attempt before resampling the unchanged request. // Its usage is carried into the next step-finish part. yield* Effect.forEach( - parts, + parts.filter((part) => !keep?.has(part.id)), (part) => session.removePart({ sessionID: part.sessionID, @@ -702,6 +702,19 @@ const layer = Layer.effect( "session.id": input.sessionID, messageID: input.assistantMessage.id, }) + const retained = new Set( + (yield* MessageV2.parts(ctx.assistantMessage.id).pipe(Effect.provideService(Database.Service, database))).map( + (part) => part.id, + ), + ) + let toolExecuted = false + const request = { + ...streamInput, + onToolExecution: () => { + toolExecuted = true + streamInput.onToolExecution?.() + }, + } ctx.needsCompaction = false ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true @@ -710,10 +723,13 @@ const layer = Layer.effect( ctx.currentText = undefined ctx.reasoningMap = {} yield* status.set(ctx.sessionID, { type: "busy" }) - const stream = llm.stream(streamInput) + const stream = llm.stream(request) yield* stream.pipe( - Stream.tap((event) => handleEvent(event)), + Stream.tap((event) => { + if (event.type === "tool-call" || event.type === "tool-result") toolExecuted = true + return handleEvent(event) + }), Stream.takeUntil(() => ctx.needsCompaction), Stream.runDrain, ) @@ -734,10 +750,15 @@ const layer = Layer.effect( SessionRetry.policy({ provider: input.model.providerID, parse, + canRetry: (error) => !SessionRetry.isTransportError(error) || !toolExecuted, // Only replace attempts that will be retried. Cloud intentionally // returns the terminal partial next to the truncation error. onRetry: (error) => - SessionV1.OutputLengthError.isInstance(error) ? resetOutputLimit() : Effect.void, + SessionV1.OutputLengthError.isInstance(error) + ? resetAttempt() + : SessionRetry.isTransportError(error) + ? resetAttempt(retained) + : Effect.void, set: (info) => { return status.set(ctx.sessionID, { type: "retry", diff --git a/packages/opencode/src/session/retry.ts b/packages/opencode/src/session/retry.ts index b10e6ad5c1..c6fc74d07c 100644 --- a/packages/opencode/src/session/retry.ts +++ b/packages/opencode/src/session/retry.ts @@ -80,6 +80,10 @@ export function delay(attempt: number, error?: SessionV1.APIError) { return cap(Math.min(RETRY_INITIAL_DELAY * Math.pow(RETRY_BACKOFF_FACTOR, attempt - 1), RETRY_MAX_DELAY_NO_HEADERS)) } +export function isTransportError(error: Err) { + return SessionV1.APIError.isInstance(error) && error.data.metadata?.code === "upstream_stream_failed" +} + export function retryable(error: Err, provider: string) { if (SessionV1.OutputLengthError.isInstance(error)) { return { message: "Model hit its output limit", maxTotalAttempts: 3 } @@ -87,6 +91,7 @@ export function retryable(error: Err, provider: string) { // context overflow errors should not be retried if (SessionV1.ContextOverflowError.isInstance(error)) return undefined if (SessionV1.APIError.isInstance(error)) { + if (isTransportError(error)) return { message: error.data.message, maxTotalAttempts: 3 } const status = error.data.statusCode // 5xx errors are transient server failures and should always be retried, // even when the provider SDK doesn't explicitly mark them as retryable. @@ -194,6 +199,7 @@ function parseJSON(value: unknown) { export function policy(opts: { provider: string parse: (error: unknown) => Err + canRetry?: (error: Err) => boolean onRetry?: (error: Err) => Effect.Effect set: (input: { attempt: number; message: string; action?: Retryable["action"]; next: number }) => Effect.Effect }) { @@ -201,7 +207,7 @@ export function policy(opts: { Effect.succeed((meta: Schedule.InputMetadata) => { const error = opts.parse(meta.input) const retry = retryable(error, opts.provider) - if (!retry) return Cause.done(meta.attempt) + if (!retry || opts.canRetry?.(error) === false) return Cause.done(meta.attempt) // `meta.attempt` is 1-based and shared by every reason (Schedule keeps one // counter per request), which is why the cap reads as a total. if (retry.maxTotalAttempts !== undefined && meta.attempt >= retry.maxTotalAttempts) diff --git a/packages/opencode/test/lib/llm-server.ts b/packages/opencode/test/lib/llm-server.ts index 245acc7280..9d29fc5bab 100644 --- a/packages/opencode/test/lib/llm-server.ts +++ b/packages/opencode/test/lib/llm-server.ts @@ -330,6 +330,7 @@ function flow(item: Sse) { } function responses(item: Sse, model: string) { + if ([...item.head, ...item.tail].some((part) => part && typeof part === "object" && "type" in part)) return item let seq = 1 let msg: string | undefined let reason: string | undefined diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 09d108bf17..2c562702ca 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -19,7 +19,7 @@ import { SessionStatus } from "../../src/session/status" import { SessionSummary } from "../../src/session/summary" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { provideTmpdirInstance, provideTmpdirServer } from "../fixture/fixture" -import { testEffect } from "../lib/effect" +import { pollWithTimeout, testEffect } from "../lib/effect" import { raw, reply, TestLLMServer } from "../lib/llm-server" import { RuntimeFlags } from "@/effect/runtime-flags" import { ProviderV2 } from "@opencode-ai/core/provider" @@ -1171,3 +1171,215 @@ itFragmentFailure.live("session.processor effect tests retain partial legacy par { config: cfg }, ), ) + +for (const mode of ["recover", "protocol", "connection", "exhaust", "executed", "permanent", "cancel"] as const) { + it.live( + `gateway transport recovery ${mode}`, + () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + yield* Effect.promise(() => Bun.write(path.join(dir, "bcode.json"), JSON.stringify(providerCfg(llm.url)))) + const { processors, session, provider } = yield* boot() + const recover = mode === "recover" || mode === "protocol" || mode === "connection" + const action = defer() + let executed = 0 + const created = { + type: "response.created", + sequence_number: 1, + response: { id: "resp_transport", created_at: 1, model: "test-model", service_tier: null }, + } + const call = { + type: "response.output_item.added", + sequence_number: 2, + output_index: 0, + item: { + type: "function_call", + id: "fc_transport", + call_id: "call_transport", + name: "lookup", + arguments: "", + }, + } + const args = { + type: "response.function_call_arguments.delta", + sequence_number: 3, + output_index: 0, + item_id: "fc_transport", + delta: '{"query":"weather"}', + } + const called = { + type: "response.output_item.done", + sequence_number: 4, + output_index: 0, + item: { ...call.item, arguments: args.delta, status: "completed" }, + } + const completed = { + type: "response.completed", + sequence_number: 9, + response: { + incomplete_details: null, + service_tier: null, + usage: { + input_tokens: 7, + input_tokens_details: { cached_tokens: null }, + output_tokens: 11, + output_tokens_details: { reasoning_tokens: null }, + }, + }, + } + const error = { + type: "error", + sequence_number: 8, + error: { + type: "upstream_error", + code: "upstream_stream_failed", + message: + mode === "permanent" + ? "ValueError: invalid local state" + : mode === "protocol" + ? "RemoteProtocolError: incomplete chunked read" + : mode === "connection" + ? "APIConnectionError: Connection error." + : "ReadError", + }, + } + const partial = [ + created, + { + type: "response.output_item.added", + sequence_number: 2, + output_index: 0, + item: { type: "reasoning", id: "rs_transport", encrypted_content: null }, + }, + { + type: "response.reasoning_summary_part.added", + sequence_number: 3, + item_id: "rs_transport", + summary_index: 0, + }, + { + type: "response.reasoning_summary_text.delta", + sequence_number: 4, + item_id: "rs_transport", + summary_index: 0, + delta: "unfinished reasoning", + }, + { + type: "response.output_item.added", + sequence_number: 5, + output_index: 1, + item: { type: "message", id: "msg_partial" }, + }, + { + type: "response.output_text.delta", + sequence_number: 6, + item_id: "msg_partial", + delta: "unfinished answer", + logprobs: null, + }, + ] + const first = + mode === "executed" + ? raw({ head: [created, call, args, called], wait: action.promise, tail: [error] }) + : raw({ chunks: [...partial, error] }) + yield* llm.push(first) + if (mode === "exhaust") { + yield* llm.push(first) + yield* llm.push(first) + } + yield* llm.push( + raw({ + chunks: [ + created, + { + type: "response.output_item.added", + sequence_number: 2, + output_index: 0, + item: { type: "message", id: "msg_complete" }, + }, + { + type: "response.output_text.delta", + sequence_number: 3, + item_id: "msg_complete", + delta: "recovered", + logprobs: null, + }, + { + type: "response.output_item.done", + sequence_number: 4, + output_index: 0, + item: { type: "message", id: "msg_complete" }, + }, + call, + args, + called, + completed, + ], + }), + ) + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "recover this turn") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const configured = yield* provider.getModel(ref.providerID, ref.modelID) + const mdl = { + ...configured, + cost: { ...configured.cost, input: 1, output: 1 }, + api: { ...configured.api, npm: "@ai-sdk/openai" }, + } + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl }) + const work = handle.process({ + user: parent, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "recover this turn" }], + tools: { + lookup: tool({ + inputSchema: z.object({ query: z.string() }), + execute: async () => { + executed += 1 + action.resolve() + return { output: "done", title: "lookup", metadata: {} } + }, + }), + }, + }) + if (mode === "cancel") { + const fiber = yield* work.pipe(Effect.forkChild) + const status = yield* SessionStatus.Service + yield* pollWithTimeout( + status.get(chat.id).pipe(Effect.map((value) => (value.type === "retry" ? true : undefined))), + "transport retry never entered backoff", + ) + yield* Fiber.interrupt(fiber) + expect(Exit.isFailure(yield* Fiber.await(fiber))).toBe(true) + expect(yield* llm.calls).toBe(1) + expect(executed).toBe(0) + return + } + const value = yield* work + const parts = yield* MessageV2.parts(msg.id) + expect(yield* llm.calls).toBe(recover ? 2 : mode === "exhaust" ? 3 : 1) + expect(value).toBe(recover ? "continue" : "stop") + expect(executed).toBe(recover || mode === "executed" ? 1 : 0) + if (recover) { + expect(handle.message.error).toBeUndefined() + expect(parts.filter((part) => part.type === "text").map((part) => part.text)).toStrictEqual(["recovered"]) + expect(parts.filter((part) => part.type === "reasoning")).toHaveLength(0) + expect(handle.message.tokens.input).toBe(7) + expect(handle.message.tokens.output).toBe(11) + expect(handle.message.cost).toBeCloseTo(18 / 1_000_000) + } else { + expect(handle.message.error).toBeDefined() + if (mode === "executed") { + expect(parts.filter((part) => part.type === "tool")).toHaveLength(1) + } + } + }), + { config: (url) => providerCfg(url) }, + ), + 20000, + ) +} From 625c05b3e044e5b42c68b9f4f80b1c1ad18073b5 Mon Sep 17 00:00:00 2001 From: Sarath S Menon Date: Wed, 7 Oct 2026 13:30:40 +0000 Subject: [PATCH 2/2] test(session): correct recovery stream fixture --- AGENTS.md | 2 +- docs/verification/v4-llm-transport-recovery.md | 4 ++-- packages/opencode/test/session/processor-effect.test.ts | 6 +++--- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a881621935..9f24f84bb0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -234,4 +234,4 @@ runs this filtered command. - Keep EventV2 replay owner claims separate from clustered Session execution ownership. - Keep the System Context algebra, registry, and built-ins in `src/system-context`; keep Context Source producers with their observed domains, and keep Session History selection plus Context Epoch persistence Session-owned. -- Retry streamed transport failures with a finite budget before tool execution starts; preserve earlier parts and discard only the incomplete attempt. +- Retry streamed transport failures with a finite budget only while no tool call has been observed and no tool execution has begun; preserve earlier parts and discard only the incomplete attempt. diff --git a/docs/verification/v4-llm-transport-recovery.md b/docs/verification/v4-llm-transport-recovery.md index 05151398e0..d8bff529ee 100644 --- a/docs/verification/v4-llm-transport-recovery.md +++ b/docs/verification/v4-llm-transport-recovery.md @@ -2,9 +2,9 @@ The Cloud proxy emits `upstream_stream_failed` after an interrupted upstream stream. BrowserCode previously treated this envelope as an unknown terminal error. The fix recognizes transport exception classes, permits at most three total attempts with existing backoff, removes only incomplete attempt parts, and refuses replay once prepared tool execution starts or a tool call is observed. -The focused test uses the actual OpenAI SDK and an HTTP 200 SSE fixture. It covers ReadError, RemoteProtocolError, APIConnectionError, exhaustion, an executed action, a non-transport exception, and interruption during backoff. Recovery requires two HTTP calls, one tool execution, recovered output only, and correct token/cost accounting. Exhaustion requires exactly three HTTP calls. +The focused test uses monotonic Responses sequence numbers, distinct output indexes, the actual OpenAI SDK and an HTTP 200 SSE fixture. It covers ReadError, RemoteProtocolError, APIConnectionError, exhaustion, an executed action, a non-transport exception, and interruption during backoff. Recovery requires two HTTP calls, one tool execution, recovered output only, and correct token/cost accounting. Exhaustion requires exactly three HTTP calls. -With the final unchanged regression (SHA256 `2caa9604b5ce2aaa90b5449e0e74a4c0411e08a9fd81c915955dcff3e1a67db3`), baseline main had two passing safety cases and five failures; the fix passed all seven. Package typecheck passed. Focused lint reported zero errors and 42 warnings. VM evidence is in `/home/exedev/v4-llm-transport-proof`. +With the final unchanged regression (SHA256 `bf17f328cead02406ab2bbd81ba2a93bb009d893efe3ce33c30b5e9d2af38673`), baseline main had two passing safety cases and five failures; the fix passed all seven. Package typecheck passed. Focused lint reported zero errors and 42 warnings. VM evidence is in `/home/exedev/v4-llm-transport-proof`. Run from `packages/opencode` with Bun 1.3.14: ```sh diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 2c562702ca..c8df1944c7 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -1311,9 +1311,9 @@ for (const mode of ["recover", "protocol", "connection", "exhaust", "executed", output_index: 0, item: { type: "message", id: "msg_complete" }, }, - call, - args, - called, + { ...call, sequence_number: 5, output_index: 1 }, + { ...args, sequence_number: 6, output_index: 1 }, + { ...called, sequence_number: 7, output_index: 1 }, completed, ], }),