Merge pull request #175 from browser-use/transport-recovery

fix(session): recover interrupted V4 model streams safely

Author: sarath-menon

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 only while no tool call has been observed and no tool execution has begun; preserve earlier parts and discard only the incomplete attempt.

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 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 `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
+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.

packages/opencode/src/provider/error.ts

@@ -97,6 +97,7 @@ export type ParsedStreamError =
       message: string
       isRetryable: boolean
       responseBody: string
+      metadata?: Record<string, string>
     }
 
 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 {

packages/opencode/src/session/llm.ts

@@ -44,6 +44,7 @@ export type StreamInput = {
   messages: ModelMessage[]
   small?: boolean
   tools: Record<string, Tool>
+  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<NonNullable<typeof value.execute>>) => {
+                  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,

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,

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<PartID>) {
         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",

packages/opencode/src/session/retry.ts

@@ -80,13 +80,18 @@ 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 }
   }
   // 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,14 +199,15 @@ function parseJSON(value: unknown) {
 export function policy(opts: {
   provider: string
   parse: (error: unknown) => Err
+  canRetry?: (error: Err) => boolean
   onRetry?: (error: Err) => Effect.Effect<void>
   set: (input: { attempt: number; message: string; action?: Retryable["action"]; next: number }) => Effect.Effect<void>
 }) {
   return Schedule.fromStepWithMetadata(
     Effect.succeed((meta: Schedule.InputMetadata<unknown>) => {
       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)

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