fix(session): bound transient transport failures

Author: literally-dan

packages/core/src/v1/config/provider.ts

@@ -108,9 +108,14 @@ export const Info = Schema.Struct({
 description:
 "Timeout in milliseconds to wait for response headers. Provider integrations may set defaults. Set to false to disable timeout.",
 }),
-chunkTimeout: Schema.optional(PositiveInt).annotate({
+chunkTimeout: Schema.optional(
+Schema.Union([PositiveInt,Schema.Literal(false)]).annotate({
+description:
+"Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted. Default is 120000 (2 minutes). Set to false to disable.",
+}),
+).annotate({
 description:
-"Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted.",
+"Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted. Default is 120000 (2 minutes). Set to false to disable.",
 }),
 }),
 [Schema.Record(Schema.String,Schema.Any)],

packages/opencode/src/provider/provider.ts

@@ -34,6 +34,16 @@ import { ProviderError } from "./error"
 
 constOPENAI_HEADER_TIMEOUT_DEFAULT=300_000
 
+// Default between-chunk timeout for streamed SSE responses. Without this,
+// a TCP socket that goes silent (wifi hiccup, NAT eviction, server crash
+// with no FIN/RST) parks the AI SDK's fullStream forever — Linux TCP
+// keepalive defaults to ~2 hours of idle before probing. 120s is short
+// enough to surface a freeze quickly, long enough that even slow
+// reasoning models emit *something* (token, ping, thinking chunk) in
+// between. Per-provider `chunkTimeout` overrides this; set to `false` to
+// disable entirely for providers with legitimately silent long pauses.
+constDEFAULT_CHUNK_TIMEOUT_MS=120_000
+
 functionwrapSSE(res: Response,ms: number,ctl: AbortController){
 if(typeofms!=="number"||ms<=0)returnres
 if(!res.body)returnres
@@ -1730,15 +1740,23 @@ const layer = Layer.effect(
 if(existing)returnexisting
 
 constcustomFetch=options["fetch"]
-constchunkTimeout=options["chunkTimeout"]
+constconfiguredChunkTimeout=options["chunkTimeout"]
 constheaderTimeout=options["headerTimeout"]
 deleteoptions["chunkTimeout"]
 deleteoptions["headerTimeout"]
+// Default to DEFAULT_CHUNK_TIMEOUT_MS unless the provider config
+// explicitly sets a value (number) or disables it (false).
+constchunkTimeout=
+configuredChunkTimeout===false
+ ? 0
+ : typeofconfiguredChunkTimeout==="number"
+ ? configuredChunkTimeout
+ : DEFAULT_CHUNK_TIMEOUT_MS
 
 options["fetch"]=async(input: any,init?: BunFetchRequestInit)=>{
 constfetchFn=customFetch??fetch
 constopts=init??{}
-constchunkAbortCtl=typeofchunkTimeout==="number"&&chunkTimeout>0 ? newAbortController() : undefined
+constchunkAbortCtl=chunkTimeout>0 ? newAbortController() : undefined
 constheaderTimeoutMs=headerTimeout===false ? undefined : headerTimeout
 constheaderTimeoutCtl=typeofheaderTimeoutMs==="number" ? timeoutController(headerTimeoutMs) : undefined
 constsignals: AbortSignal[]=[]

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

@@ -600,17 +600,78 @@ export function latest(msgs: WithParts[]) {
 return{ user, assistant, finished, tasks }
 }
 
+// Transport failures that can recover without changing the request. Keep
+// permanent endpoint failures such as ENOTFOUND and ECONNREFUSED out of this
+// list so a bad provider configuration surfaces immediately.
+constTRANSIENT_SYS_CODES=newSet([
+"ECONNRESET",
+"ETIMEDOUT",
+"EAI_AGAIN",
+"EHOSTUNREACH",
+"ENETUNREACH",
+"EPIPE",
+"UND_ERR_CONNECT_TIMEOUT",
+"UND_ERR_HEADERS_TIMEOUT",
+"UND_ERR_BODY_TIMEOUT",
+"UND_ERR_SOCKET",
+])
+constPERMANENT_SYS_CODES=newSet(["ENOTFOUND","ECONNREFUSED"])
+constCAUSE_MAX_DEPTH=8
+
+exportfunctionisPermanentTransportCode(code: unknown){
+returntypeofcode==="string"&&PERMANENT_SYS_CODES.has(code)
+}
+
+// Exact messages emitted for a prematurely closed transport. Generic text
+// such as "network error" or "fetch failed" cannot distinguish a transient
+// disconnect from a permanent DNS, TLS, authentication, or endpoint failure.
+constTRANSIENT_MESSAGES=newSet(["socket hang up","sse read timed out","other side closed","terminated"])
+
+functionclassifyTransportError(error: unknown){
+constseen=newSet<Error>()
+letcurrent=error
+lettransient: SystemError|undefined
+for(letdepth=0;depth<CAUSE_MAX_DEPTH;depth++){
+if(!(currentinstanceofError)||seen.has(current))break
+seen.add(current)
+constcode="code"incurrent&&typeofcurrent.code==="string" ? current.code : undefined
+if(isPermanentTransportCode(code))return{type: "permanent"asconst,error: currentasSystemError}
+if(!transient&&code&&TRANSIENT_SYS_CODES.has(code))transient=currentasSystemError
+current=current.cause
+}
+if(transient)return{type: "transient"asconst,error: transient}
+returnundefined
+}
+
 exportfunctionfromError(
 e: unknown,
 ctx: {providerID: ProviderV2.ID;aborted?: boolean},
 ): NonNullable<Assistant["error"]>{
+consttransport=classifyTransportError(e)
+consttransient=transport?.type==="transient" ? transport.error : undefined
+constpermanent=transport?.type==="permanent" ? transport.error : undefined
+constsysCode=transient?.code
 switch(true){
 caseeinstanceofDOMException&&e.name==="AbortError":
-returnnewAbortedError(
-{message: e.message},
-{
-cause: e,
-},
+// A *bare* AbortError only ever originates from `controller.abort()`
+// with no reason — i.e. a user/parent cancel (see provider.ts fetch
+// wrapper + processor onInterrupt). Every transport timeout we control
+// aborts with a *specific* reason and therefore surfaces as its own
+// identifiable error instead: AbortSignal.timeout() -> "TimeoutError"
+// DOMException (next case), the header-timeout -> HeaderTimeoutError,
+// and the SSE chunk-timeout -> ResponseStreamError ("SSE read timed
+// out"). Those are all classified retryable on their own below, so we
+// do NOT need to second-guess a bare AbortError here. Classifying by
+// error identity (rather than the externally-set `ctx.aborted` flag,
+// which races the abort propagating through the failure channel) keeps
+// a genuine cancel from being retried.
+returnnewAbortedError({message: e.message},{cause: e}).toObject()
+caseeinstanceofDOMException&&e.name==="TimeoutError":
+// Modern fetch surfaces AbortSignal.timeout() as a DOMException with
+// name "TimeoutError" rather than "AbortError". Always retryable.
+returnnewAPIError(
+{message: e.message||"Request timed out",isRetryable: true,metadata: {code: "TimeoutError"}},
+{cause: e},
 ).toObject()
 caseOutputLengthError.isInstance(e):
 returne
@@ -622,15 +683,15 @@ export function fromError(
 },
 {cause: e},
 ).toObject()
-case(easSystemError)?.code==="ECONNRESET":
+casetypeofsysCode==="string"&&TRANSIENT_SYS_CODES.has(sysCode):
 returnnewAPIError(
 {
-message: "Connection reset by server",
+message: sysCode==="ECONNRESET" ? "Connection reset by server" : `Network error (${sysCode})`,
 isRetryable: true,
 metadata: {
-code: (easSystemError).code??"",
-syscall: (easSystemError).syscall??"",
-message: (easSystemError).message??"",
+code: sysCode??"",
+syscall: transient?.syscall??"",
+message: transient?.message??"",
 },
 },
 {cause: e},
@@ -692,13 +753,25 @@ export function fromError(
 {
 message: parsed.message,
 statusCode: parsed.statusCode,
-isRetryable: parsed.isRetryable,
+isRetryable: permanent ? false : parsed.isRetryable,
 responseHeaders: parsed.responseHeaders,
 responseBody: parsed.responseBody,
-metadata: parsed.metadata,
+metadata: permanent
+ ? {
+ ...parsed.metadata,
+code: permanent.code??"",
+syscall: permanent.syscall??"",
+message: permanent.message??"",
+}
+ : parsed.metadata,
 },
 {cause: e},
 ).toObject()
+caseeinstanceofError&&TRANSIENT_MESSAGES.has(e.message.toLowerCase()):
+returnnewAPIError(
+{message: e.message,isRetryable: true,metadata: {code: "NETWORK_ERROR",message: e.message}},
+{cause: e},
+).toObject()
 caseeinstanceofError:
 returnnewNamedError.Unknown({message: errorMessage(e)},{cause: e}).toObject()
 default:

packages/opencode/src/session/retry.ts

@@ -25,6 +25,7 @@ export type Retryable = {
 
 exportconstRETRY_INITIAL_DELAY=2000
 exportconstRETRY_BACKOFF_FACTOR=2
+exportconstRETRY_MAX_ATTEMPTS=5
 exportconstRETRY_MAX_DELAY_NO_HEADERS=30_000// 30 seconds
 exportconstRETRY_MAX_DELAY=2_147_483_647// max 32-bit signed integer for setTimeout
 
@@ -69,6 +70,7 @@ export function retryable(error: Err, provider: string) {
 // context overflow errors should not be retried
 if(SessionV1.ContextOverflowError.isInstance(error))returnundefined
 if(SessionV1.APIError.isInstance(error)){
+if(MessageV2.isPermanentTransportCode(error.data.metadata?.code))returnundefined
 conststatus=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.
@@ -180,6 +182,7 @@ export function policy(opts: {
 }){
 returnSchedule.fromStepWithMetadata(
 Effect.succeed((meta: Schedule.InputMetadata<unknown>)=>{
+if(meta.attempt>RETRY_MAX_ATTEMPTS)returnCause.done(meta.attempt)
 consterror=opts.parse(meta.input)
 constretry=retryable(error,opts.provider)
 if(!retry)returnCause.done(meta.attempt)

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

@@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"
 import{SessionV1}from"@opencode-ai/core/v1/session"
 import{APICallError}from"ai"
 import{MessageV2}from"../../src/session/message-v2"
+import{SessionRetry}from"../../src/session/retry"
 import{ProviderTransform}from"@/provider/transform"
 importtype{Provider}from"@/provider/provider"
 
@@ -1552,6 +1553,130 @@ describe("session.message-v2.fromError", () => {
 
 expect(result.name).toBe("MessageAbortedError")
 })
+
+test("classifies user-cancelled AbortError as AbortedError when ctx.aborted is true",()=>{
+consterr=newDOMException("Aborted","AbortError")
+constresult=MessageV2.fromError(err,{ providerID,aborted: true})
+expect(result.name).toBe("MessageAbortedError")
+})
+
+test("classifies a bare AbortError as a cancel regardless of ctx.aborted",()=>{
+// A bare AbortError only comes from controller.abort() with no reason —
+// i.e. a user/parent cancel. Transport timeouts surface as their own
+// specific errors (TimeoutError/HeaderTimeoutError/ResponseStreamError),
+// so a bare AbortError must NOT be reclassified as retryable — otherwise
+// a cancel races the failure channel and gets retried.
+consterr=newDOMException("The operation was aborted","AbortError")
+constresult=MessageV2.fromError(err,{ providerID })
+expect(result.name).toBe("MessageAbortedError")
+})
+
+test("classifies TimeoutError DOMException as retryable APIError",()=>{
+// AbortSignal.timeout() in modern fetch surfaces as TimeoutError.
+consterr=newDOMException("The operation timed out","TimeoutError")
+constresult=MessageV2.fromError(err,{ providerID })
+expect(SessionV1.APIError.isInstance(result)).toBe(true)
+expect((resultasSessionV1.APIError).data.isRetryable).toBe(true)
+})
+
+test.each([
+["ETIMEDOUT","Network error (ETIMEDOUT)"],
+["EAI_AGAIN","Network error (EAI_AGAIN)"],
+["EHOSTUNREACH","Network error (EHOSTUNREACH)"],
+["ENETUNREACH","Network error (ENETUNREACH)"],
+["EPIPE","Network error (EPIPE)"],
+["UND_ERR_CONNECT_TIMEOUT","Network error (UND_ERR_CONNECT_TIMEOUT)"],
+["UND_ERR_HEADERS_TIMEOUT","Network error (UND_ERR_HEADERS_TIMEOUT)"],
+["UND_ERR_BODY_TIMEOUT","Network error (UND_ERR_BODY_TIMEOUT)"],
+["UND_ERR_SOCKET","Network error (UND_ERR_SOCKET)"],
+])("classifies %s SystemError as retryable APIError",(code,expectedMessage)=>{
+consterr=Object.assign(newError(`${code} from test`),{ code })
+constresult=MessageV2.fromError(err,{ providerID })
+expect(SessionV1.APIError.isInstance(result)).toBe(true)
+expect((resultasSessionV1.APIError).data.isRetryable).toBe(true)
+expect((resultasSessionV1.APIError).data.message).toBe(expectedMessage)
+})
+
+test("classifies a fetch failure with a transient cause as retryable",()=>{
+constcause=Object.assign(newError("read ECONNRESET"),{code: "ECONNRESET"})
+constwrapper=Object.assign(newTypeError("fetch failed",{ cause }),{code: "ERR_FETCH_FAILED"})
+constresult=MessageV2.fromError(wrapper,{ providerID })
+expect(SessionV1.APIError.isInstance(result)).toBe(true)
+expect((resultasSessionV1.APIError).data.metadata?.code).toBe("ECONNRESET")
+})
+
+test.each([
+["ENOTFOUND","getaddrinfo"],
+["ECONNREFUSED","connect"],
+])("overrides retryable APICallError caused by %s",(code,syscall)=>{
+constcause=Object.assign(newError(`${syscall}${code} api.invalid`),{ code, syscall })
+constwrapper=Object.assign(newTypeError("fetch failed",{ cause }),{code: "ERR_FETCH_FAILED"})
+consterror=newAPICallError({
+message: "Cannot connect to API",
+url: "https://api.invalid/v1/messages",
+requestBodyValues: {model: "test"},
+cause: wrapper,
+isRetryable: true,
+})
+
+constresult=MessageV2.fromError(error,{ providerID })
+expect(SessionV1.APIError.isInstance(result)).toBe(true)
+if(!SessionV1.APIError.isInstance(result))thrownewError("expected APIError")
+expect(result.data.isRetryable).toBe(false)
+expect(result.data.metadata).toMatchObject({ code, syscall,url: "https://api.invalid/v1/messages"})
+expect(SessionRetry.retryable(result,"test")).toBeUndefined()
+})
+
+test("gives a nested permanent code precedence over a transient wrapper",()=>{
+constcause=Object.assign(newError("getaddrinfo ENOTFOUND api.invalid"),{code: "ENOTFOUND"})
+constwrapper=Object.assign(newError("socket reset",{ cause }),{code: "ECONNRESET"})
+constresult=MessageV2.fromError(wrapper,{ providerID })
+expect(result.name).toBe("UnknownError")
+})
+
+test("stops traversing cyclic error causes",()=>{
+constfirst=Object.assign(newError("first wrapper"),{code: "ERR_FIRST"})
+constsecond=Object.assign(newError("second wrapper",{cause: first}),{code: "ERR_SECOND"})
+first.cause=second
+constresult=MessageV2.fromError(first,{ providerID })
+expect(result.name).toBe("UnknownError")
+})
+
+test("bounds deeply nested error causes",()=>{
+constcause=Object.assign(newError("read ECONNRESET"),{code: "ECONNRESET"})
+constwrapper=Array.from({length: 32},(_,index)=>index).reduce<Error>(
+(current,index)=>newError(`wrapper ${index}`,{cause: current}),
+cause,
+)
+constresult=MessageV2.fromError(wrapper,{ providerID })
+expect(result.name).toBe("UnknownError")
+})
+
+test.each(["socket hang up","SSE read timed out","other side closed","terminated"])(
+"classifies '%s' bare Error message as retryable APIError",
+(message)=>{
+constresult=MessageV2.fromError(newError(message),{ providerID })
+expect(SessionV1.APIError.isInstance(result)).toBe(true)
+expect((resultasSessionV1.APIError).data.isRetryable).toBe(true)
+},
+)
+
+test.each([
+"fetch failed",
+"Failed to fetch account configuration",
+"network error: certificate has expired",
+"Network request failed because access is forbidden",
+"connect error: invalid provider base URL",
+"request terminated because the API key is invalid",
+])("does not treat '%s' as a transient transport error",(message)=>{
+constresult=MessageV2.fromError(newError(message),{ providerID })
+expect(result.name).toBe("UnknownError")
+})
+
+test("leaves unrelated Error messages classified as Unknown",()=>{
+constresult=MessageV2.fromError(newError("Some unrelated bug"),{ providerID })
+expect(result.name).toBe("UnknownError")
+})
 })
 
 describe("session.message-v2.latest",()=>{