fix(opencode): bound codex stream stalls

Author: avilabss

packages/opencode/src/plugin/codex.ts

@@ -11,6 +11,8 @@ const log = Log.create({ service: "plugin.codex" })
 const CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
 const ISSUER = "https://auth.openai.com"
 const CODEX_API_ENDPOINT = "https://chatgpt.com/backend-api/codex/responses"
+const CODEX_REQUEST_TIMEOUT = 30_000
+const CODEX_CHUNK_TIMEOUT = 120_000
 const OAUTH_PORT = 1455
 const OAUTH_POLLING_SAFETY_MARGIN_MS = 3000
 const ALLOWED_MODELS = new Set([
@@ -140,9 +142,10 @@ async function exchangeCodeForTokens(code: string, redirectUri: string, pkce: Pk
   return response.json()
 }
 
-async function refreshAccessToken(refreshToken: string, issuer = ISSUER): Promise<TokenResponse> {
+async function refreshAccessToken(refreshToken: string, issuer = ISSUER, signal?: AbortSignal): Promise<TokenResponse> {
   const response = await fetch(`${issuer}/oauth/token`, {
     method: "POST",
+    signal,
     headers: { "Content-Type": "application/x-www-form-urlencoded" },
     body: new URLSearchParams({
       grant_type: "refresh_token",
@@ -421,6 +424,8 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
 
         return {
           apiKey: OAUTH_DUMMY_KEY,
+          timeout: CODEX_REQUEST_TIMEOUT,
+          chunkTimeout: CODEX_CHUNK_TIMEOUT,
           async fetch(requestInput: RequestInfo | URL, init?: RequestInit) {
             // Remove dummy API key authorization header
             if (init?.headers) {
@@ -445,7 +450,7 @@ export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPlug
             if (!currentAuth.access || currentAuth.expires < Date.now()) {
               if (!refreshPromise) {
                 log.info("refreshing codex access token")
-                refreshPromise = refreshAccessToken(currentAuth.refresh, issuer)
+                refreshPromise = refreshAccessToken(currentAuth.refresh, issuer, init?.signal ?? undefined)
                   .then(async (tokens) => {
                     const accountId = extractAccountId(tokens) || authWithAccount.accountId
                     await input.client.auth.set({

packages/opencode/src/provider/provider.ts

@@ -30,6 +30,7 @@ import { ModelStatus } from "./model-status"
 import { RuntimeFlags } from "@/effect/runtime-flags"
 
 const log = Log.create({ service: "provider" })
+const PROVIDER_TIMEOUT_DEFAULT = 300_000
 
 function shouldUseCopilotResponsesApi(modelID: string): boolean {
   const match = /^gpt-(\d+)/.exec(modelID)
@@ -85,6 +86,18 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
   })
 }
 
+function timeoutController(ms: number) {
+  const ctl = new AbortController()
+  const id = setTimeout(
+    () => ctl.abort(new DOMException(`Provider request timed out after ${ms}ms`, "TimeoutError")),
+    ms,
+  )
+  return {
+    signal: ctl.signal,
+    clear: () => clearTimeout(id),
+  }
+}
+
 function googleVertexAnthropicBaseURL(project: string | undefined, location: string | undefined) {
   if (!project) return
   if (location !== "eu" && location !== "us") return
@@ -1607,12 +1620,13 @@ export const layer = Layer.effect(
           const fetchFn = customFetch ?? fetch
           const opts = init ?? {}
           const chunkAbortCtl = typeof chunkTimeout === "number" && chunkTimeout > 0 ? new AbortController() : undefined
+          const requestTimeout = options["timeout"] === false ? undefined : (options["timeout"] ?? PROVIDER_TIMEOUT_DEFAULT)
+          const requestTimeoutCtl = typeof requestTimeout === "number" ? timeoutController(requestTimeout) : undefined
           const signals: AbortSignal[] = []
 
           if (opts.signal) signals.push(opts.signal)
           if (chunkAbortCtl) signals.push(chunkAbortCtl.signal)
-          if (options["timeout"] !== undefined && options["timeout"] !== null && options["timeout"] !== false)
-            signals.push(AbortSignal.timeout(options["timeout"]))
+          if (requestTimeoutCtl) signals.push(requestTimeoutCtl.signal)
 
           const combined = signals.length === 0 ? null : signals.length === 1 ? signals[0] : AbortSignal.any(signals)
           if (combined) opts.signal = combined
@@ -1639,7 +1653,7 @@ export const layer = Layer.effect(
             ...opts,
             // @ts-ignore see here: https://github.com/oven-sh/bun/issues/16682
             timeout: false,
-          })
+          }).finally(() => requestTimeoutCtl?.clear())
 
           if (!chunkAbortCtl) return res
           return wrapSSE(res, chunkTimeout, chunkAbortCtl)

packages/opencode/src/session/retry.ts

@@ -125,6 +125,9 @@ export function retryable(error: Err, provider: string) {
   const msg = isRecord(error.data) ? error.data.message : undefined
   if (typeof msg === "string") {
     const lower = msg.toLowerCase()
+    if (lower.includes("sse read timed out") || lower.includes("provider request timed out")) {
+      return { message: msg }
+    }
     if (
       lower.includes("rate increased too quickly") ||
       lower.includes("rate limit") ||

packages/opencode/test/session/retry.test.ts

@@ -163,6 +163,14 @@ describe("session.retry.retryable", () => {
     expect(SessionRetry.retryable(error, retryProvider)).toEqual({ message: msg })
   })
 
+  test("retries transport timeout errors", () => {
+    const sse = wrap("SSE read timed out")
+    expect(SessionRetry.retryable(sse, retryProvider)).toEqual({ message: "SSE read timed out" })
+
+    const request = wrap("Provider request timed out after 30000ms")
+    expect(SessionRetry.retryable(request, retryProvider)).toEqual({ message: "Provider request timed out after 30000ms" })
+  })
+
   test("does not retry context overflow errors", () => {
     const error = new MessageV2.ContextOverflowError({
       message: "Input exceeds context window of this model",