fix(proxy): timeout stalled OpenAI-compatible streams (#548)

Guard OpenAI-compatible proxy streaming bodies with the configured AI request timeout so a provider that emits partial SSE and then idles cannot leave proxy consumers waiting forever.

This is a proxy-level fix found while investigating #548; it does not claim to close the broader desktop interruption issue.

Tested: bun test src/server/__tests__/proxy-network-settings.test.ts

Tested: bun run check:server

Confidence: medium

Scope-risk: narrow

src/server/__tests__/proxy-network-settings.test.ts

@@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test'
 import * as fs from 'fs/promises'
 import * as os from 'os'
 import * as path from 'path'
-import { handleProxyRequest } from '../proxy/handler.js'
+import { handleProxyRequest, withStreamIdleTimeout } from '../proxy/handler.js'
 import { ProviderService } from '../services/providerService.js'
 import { resetSettingsCache } from '../../utils/settings/settingsCache.js'
 
@@ -180,7 +180,7 @@ describe('proxy network settings', () => {
     }
   })
 
-  test('uses configured AI request timeout only while opening streaming upstream requests', async () => {
+  test('uses configured AI request timeout while opening and reading streaming upstream requests', async () => {
     await fs.writeFile(
       path.join(tmpDir, 'settings.json'),
       JSON.stringify({
@@ -262,7 +262,11 @@ describe('proxy network settings', () => {
 
       expect(res.status).toBe(200)
       expect(timeoutCalls).toEqual([])
-      expect(timers).toEqual([{ ms: 180_000, cleared: true }])
+      expect(timers).toEqual([
+        { ms: 180_000, cleared: true },
+        { ms: 180_000, cleared: true },
+        { ms: 180_000, cleared: true },
+      ])
     } finally {
       globalThis.clearTimeout = originalClearTimeout
       globalThis.setTimeout = originalSetTimeout
@@ -270,4 +274,55 @@ describe('proxy network settings', () => {
       globalThis.fetch = originalFetch
     }
   })
+
+  test('fails a streaming upstream body that stops producing chunks', async () => {
+    const stalled = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n'))
+      },
+    })
+
+    await expect(new Response(withStreamIdleTimeout(stalled, 20)).text())
+      .rejects
+      .toThrow('Upstream stream idle timeout after 20ms')
+  })
+
+  test('propagates streaming upstream body errors before the idle timeout fires', async () => {
+    let pulls = 0
+    const upstream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pulls += 1
+        if (pulls === 1) {
+          controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n'))
+          return
+        }
+        controller.error(new Error('upstream body failed'))
+      },
+    })
+    const reader = withStreamIdleTimeout(upstream, 1_000).getReader()
+
+    expect(await reader.read()).toEqual({
+      done: false,
+      value: new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n'),
+    })
+    await expect(reader.read()).rejects.toThrow('upstream body failed')
+  })
+
+  test('cancels the upstream body when the downstream stream is canceled', async () => {
+    let cancelReason: unknown = null
+    const upstream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n'))
+      },
+      cancel(reason) {
+        cancelReason = reason
+      },
+    })
+    const reader = withStreamIdleTimeout(upstream, 1_000).getReader()
+
+    expect((await reader.read()).done).toBe(false)
+    await reader.cancel('downstream closed')
+
+    expect(cancelReason).toBe('downstream closed')
+  })
 })

src/server/proxy/handler.ts

@@ -69,6 +69,58 @@ async function fetchUpstreamWithTimeout(
   }
 }
 
+export function withStreamIdleTimeout(
+  upstream: ReadableStream<Uint8Array>,
+  timeoutMs: number,
+): ReadableStream<Uint8Array> {
+  let reader: ReadableStreamDefaultReader<Uint8Array> | null = null
+  let timer: ReturnType<typeof setTimeout> | null = null
+
+  const clearIdleTimer = () => {
+    if (timer) {
+      clearTimeout(timer)
+      timer = null
+    }
+  }
+
+  return new ReadableStream({
+    async start(controller) {
+      reader = upstream.getReader()
+      let timedOut = false
+
+      const armIdleTimer = () => {
+        clearIdleTimer()
+        timer = setTimeout(() => {
+          timedOut = true
+          void reader?.cancel('stream idle timeout').catch(() => undefined)
+          controller.error(new Error(`Upstream stream idle timeout after ${timeoutMs}ms`))
+        }, timeoutMs)
+      }
+
+      try {
+        armIdleTimer()
+        while (true) {
+          const { done, value } = await reader.read()
+          if (done) break
+          if (timedOut) break
+
+          controller.enqueue(value)
+          armIdleTimer()
+        }
+        clearIdleTimer()
+        if (!timedOut) controller.close()
+      } catch (err) {
+        clearIdleTimer()
+        if (!timedOut) controller.error(err)
+      }
+    },
+    cancel(reason) {
+      clearIdleTimer()
+      return reader?.cancel(reason)
+    },
+  })
+}
+
 export async function handleProxyRequest(req: Request, url: URL): Promise<Response> {
   const providerMatch = url.pathname.match(/^\/proxy\/providers\/([^/]+)\/v1\/messages$/)
   const providerId = providerMatch ? decodeURIComponent(providerMatch[1]!) : undefined
@@ -207,7 +259,8 @@ async function handleOpenaiChat(
         { status: 502 },
       )
     }
-    const anthropicStream = openaiChatStreamToAnthropic(upstream.body, body.model)
+    const upstreamBody = withStreamIdleTimeout(upstream.body, aiRequestTimeoutMs)
+    const anthropicStream = openaiChatStreamToAnthropic(upstreamBody, body.model)
     return new Response(anthropicStream, {
       status: 200,
       headers: {
@@ -278,7 +331,8 @@ async function handleOpenaiResponses(
         { status: 502 },
       )
     }
-    const anthropicStream = openaiResponsesStreamToAnthropic(upstream.body, body.model)
+    const upstreamBody = withStreamIdleTimeout(upstream.body, aiRequestTimeoutMs)
+    const anthropicStream = openaiResponsesStreamToAnthropic(upstreamBody, body.model)
     return new Response(anthropicStream, {
       status: 200,
       headers: {