fix(events): clear stale pending questions from remote servers

backend/src/routes/opencode-server-proxy.ts

@@ -12,6 +12,7 @@ import { logger } from '../utils/logger'
 const MOUNT_PREFIX = '/api/opencode/servers/'
 const REQUEST_EXCLUDED_HEADERS = new Set(['cookie', 'origin', 'referer'])
 const RESPONSE_EXCLUDED_HEADERS = new Set(['set-cookie'])
+const UPSTREAM_HEADERS_TIMEOUT_MS = 30_000
 
 type UpstreamPath = { ok: true; path: string } | { ok: false; status: 400 | 404 }
 
@@ -82,6 +83,7 @@ export function createOpenCodeServerProxyRoutes(
   registry: OpenCodeServerRegistry,
   requireAuth: MiddlewareHandler,
   fetchFn: typeof fetch = fetch,
+  upstreamHeadersTimeoutMs: number = UPSTREAM_HEADERS_TIMEOUT_MS,
 ): Hono {
   const app = new Hono()
 
@@ -117,16 +119,33 @@ export function createOpenCodeServerProxyRoutes(
 
     const hasBody = c.req.method !== 'GET' && c.req.method !== 'HEAD'
 
+    let headersTimeoutTimer: ReturnType<typeof setTimeout> | null = null
+    let headersTimedOut = false
+    let upstreamSignal = c.req.raw.signal
+    if (!hasBody) {
+      const timeoutController = new AbortController()
+      headersTimeoutTimer = setTimeout(() => {
+        headersTimedOut = true
+        timeoutController.abort()
+      }, upstreamHeadersTimeoutMs)
+      upstreamSignal = AbortSignal.any([c.req.raw.signal, timeoutController.signal])
+    }
+
     try {
       const upstreamResponse = await fetchFn(upstream.url.toString(), {
         method: c.req.method,
         headers,
         body: hasBody ? c.req.raw.body : undefined,
         redirect: 'manual',
-        signal: c.req.raw.signal,
+        signal: upstreamSignal,
         duplex: 'half',
       })
 
+      if (headersTimeoutTimer !== null) {
+        clearTimeout(headersTimeoutTimer)
+        headersTimeoutTimer = null
+      }
+
       const responseHeaders = new Headers(
         filterProxyHeaders(upstreamResponse.headers, RESPONSE_EXCLUDED_HEADERS),
       )
@@ -140,8 +159,16 @@ export function createOpenCodeServerProxyRoutes(
         headers: responseHeaders,
       })
     } catch (error) {
+      if (headersTimedOut) {
+        logger.warn(`OpenCode server proxy timed out waiting for headers: ${c.req.method} ${upstream.url.host}${upstreamPath.path}`, error)
+        return c.json({ error: 'Remote OpenCode server timed out' }, 504)
+      }
       logger.warn(`OpenCode server proxy request failed: ${c.req.method} ${upstream.url.host}${upstreamPath.path}`, error)
       return c.json({ error: 'Remote OpenCode server unreachable' }, 502)
+    } finally {
+      if (headersTimeoutTimer !== null) {
+        clearTimeout(headersTimeoutTimer)
+      }
     }
   })
 

backend/test/routes/opencode-server-proxy.test.ts

@@ -21,9 +21,13 @@ const passThroughAuth: MiddlewareHandler = async (_c, next) => {
 function buildApp(
   fetchFn: typeof fetch,
   requireAuth: MiddlewareHandler = passThroughAuth,
+  upstreamHeadersTimeoutMs?: number,
 ): Hono {
   const app = new Hono()
-  app.route('/api/opencode/servers', createOpenCodeServerProxyRoutes(registry, requireAuth, fetchFn))
+  app.route(
+    '/api/opencode/servers',
+    createOpenCodeServerProxyRoutes(registry, requireAuth, fetchFn, upstreamHeadersTimeoutMs),
+  )
   return app
 }
 
@@ -200,6 +204,65 @@ describe('opencode server proxy routes', () => {
     expect(await res.json()).toEqual({ error: 'Remote OpenCode server unreachable' })
   })
 
+  it('returns 504 when a GET upstream does not send headers before the timeout', async () => {
+    const fetchMock = vi.fn((_url: string, init: RequestInit) => new Promise<Response>((_resolve, reject) => {
+      init.signal?.addEventListener('abort', () => {
+        reject(new DOMException('Aborted', 'AbortError'))
+      })
+    }))
+    const app = buildApp(fetchMock as unknown as typeof fetch, passThroughAuth, 20)
+
+    const res = await app.request('/api/opencode/servers/srv_1/api/session')
+
+    expect(res.status).toBe(504)
+    expect(await res.json()).toEqual({ error: 'Remote OpenCode server timed out' })
+  })
+
+  it('does not time out a POST that takes longer than the headers timeout', async () => {
+    let receivedSignal: AbortSignal | null | undefined
+    const fetchMock = vi.fn(async (_url: string, init: RequestInit) => {
+      receivedSignal = init.signal
+      await new Promise((resolve) => setTimeout(resolve, 40))
+      return new Response('ok', { status: 200 })
+    })
+    const app = buildApp(fetchMock as unknown as typeof fetch, passThroughAuth, 10)
+
+    const res = await app.request('/api/opencode/servers/srv_1/api/session/ses_1/message', {
+      method: 'POST',
+      body: '{}',
+    })
+
+    expect(res.status).toBe(200)
+    expect(receivedSignal?.aborted).toBe(false)
+  })
+
+  it('keeps streaming the upstream body after headers arrive', async () => {
+    const encoder = new TextEncoder()
+    let receivedSignal: AbortSignal | null | undefined
+    const fetchMock = vi.fn(async (_url: string, init: RequestInit) => {
+      receivedSignal = init.signal
+      const stream = new ReadableStream<Uint8Array>({
+        async start(controller) {
+          controller.enqueue(encoder.encode('data: one\n\n'))
+          await new Promise((resolve) => setTimeout(resolve, 40))
+          controller.enqueue(encoder.encode('data: two\n\n'))
+          controller.close()
+        },
+      })
+      return new Response(stream, {
+        status: 200,
+        headers: { 'content-type': 'text/event-stream' },
+      })
+    })
+    const app = buildApp(fetchMock as unknown as typeof fetch, passThroughAuth, 10)
+
+    const res = await app.request('/api/opencode/servers/srv_1/api/event')
+
+    expect(res.status).toBe(200)
+    expect(await res.text()).toBe('data: one\n\ndata: two\n\n')
+    expect(receivedSignal?.aborted).toBe(false)
+  })
+
   it('streams the upstream response body and drops set-cookie', async () => {
     const encoder = new TextEncoder()
     const stream = new ReadableStream<Uint8Array>({