fix(vscode): recover stalled OpenCode SSE streams

packages/vscode/src/DOCUMENTATION.md

@@ -46,6 +46,7 @@ The webview CSP permits `blob:` only for `worker-src` so shared UI parsers can r
 
 - `bridge-proxy-runtime.ts`
   - Proxy route handlers (`api:proxy`, `api:session:message`) with injected helper dependencies.
+  - SSE routes are intentionally excluded from the generic proxy and use `sseProxy.ts`, whose upstream-only stall watchdog closes a quiet OpenCode stream so the webview can reconnect instead of trusting an open but silent response.
 
 - `bridge-config-runtime.ts`
   - Config and skills message handlers (`api:config/*`).

packages/vscode/src/sseProxy.test.ts

@@ -0,0 +1,101 @@
+import { describe, test } from 'node:test';
+import assert from 'node:assert/strict';
+import type { OpenCodeManager } from './opencode';
+import { openSseProxy } from './sseProxy';
+
+const createManager = (): OpenCodeManager => ({
+  start: async () => {},
+  stop: async () => {},
+  restart: async () => {},
+  setWorkingDirectory: async (path) => ({ success: true, path }),
+  getStatus: () => 'connected',
+  getApiUrl: () => 'http://127.0.0.1:3902',
+  getOpenCodeAuthHeaders: () => ({}),
+  getWorkingDirectory: () => '/workspace',
+  isCliAvailable: () => true,
+  getDebugInfo: () => ({
+    mode: 'managed',
+    status: 'connected',
+    workingDirectory: '/workspace',
+    cliAvailable: true,
+    cliPath: null,
+    configuredApiUrl: null,
+    configuredPort: null,
+    detectedPort: 3902,
+    apiPrefix: '',
+    apiPrefixDetected: true,
+    startCount: 1,
+    restartCount: 0,
+    lastStartAt: null,
+    lastConnectedAt: null,
+    lastExitCode: null,
+    serverUrl: 'http://127.0.0.1:3902',
+    lastReadyElapsedMs: null,
+    lastReadyAttempts: null,
+    lastStartAttempts: null,
+    version: null,
+    secureConnection: false,
+    authSource: null,
+  }),
+  onStatusChange: (callback) => {
+    callback('connected');
+    return { dispose: () => {} };
+  },
+});
+
+describe('VS Code SSE proxy', () => {
+  test('closes a quiet upstream SSE stream after the stall timeout', async () => {
+    const originalFetch = globalThis.fetch;
+
+    try {
+      globalThis.fetch = (async () => new Response(new ReadableStream<Uint8Array>({}), {
+        status: 200,
+        headers: { 'content-type': 'text/event-stream' },
+      })) as typeof fetch;
+
+      const controller = new AbortController();
+      const proxy = await openSseProxy({
+        manager: createManager(),
+        path: '/global/event',
+        signal: controller.signal,
+        stallTimeoutMs: 20,
+        onChunk: () => assert.fail('quiet stream should not emit chunks'),
+      });
+
+      await assert.doesNotReject(proxy.run);
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  });
+
+  test('resets the stall timeout when upstream bytes arrive', async () => {
+    const originalFetch = globalThis.fetch;
+
+    try {
+      globalThis.fetch = (async () => new Response(new ReadableStream<Uint8Array>({
+        start(controller) {
+          setTimeout(() => controller.enqueue(new TextEncoder().encode(':first\n\n')), 5);
+          setTimeout(() => controller.enqueue(new TextEncoder().encode('data: second\n\n')), 15);
+        },
+      }), {
+        status: 200,
+        headers: { 'content-type': 'text/event-stream' },
+      })) as typeof fetch;
+
+      const chunks: string[] = [];
+      const controller = new AbortController();
+      const proxy = await openSseProxy({
+        manager: createManager(),
+        path: '/global/event',
+        signal: controller.signal,
+        stallTimeoutMs: 18,
+        onChunk: (chunk) => chunks.push(chunk),
+      });
+
+      await assert.doesNotReject(proxy.run);
+      assert.deepEqual(chunks, [':first\n\n', 'data: second\n\n']);
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  });
+});

packages/vscode/src/sseProxy.ts

@@ -7,6 +7,7 @@ type OpenSseProxyOptions = {
   headers?: Record<string, string>;
   signal: AbortSignal;
   onChunk: (chunk: string) => void;
+  stallTimeoutMs?: number;
 };
 
 type OpenSseProxyResult = {
@@ -22,6 +23,7 @@ const SSE_RESPONSE_HEADERS = {
 // SSE reconnect configuration
 const MAX_RECONNECTS = 3;
 const BASE_RECONNECT_DELAY = 1000; // 1 second
+const DEFAULT_UPSTREAM_STALL_TIMEOUT_MS = 20000;
 
 const sleep = (ms: number, signal: AbortSignal) => new Promise<void>((resolve) => {
   if (signal.aborted) {
@@ -117,21 +119,54 @@ const fetchSseResponse = async (
   return response;
 };
 
-const pipeSseResponse = async (response: Response, signal: AbortSignal, onChunk: (chunk: string) => void): Promise<void> => {
+const resolveStallTimeoutMs = (value: number | undefined): number => (
+  Number.isFinite(value) && typeof value === 'number' ? value : DEFAULT_UPSTREAM_STALL_TIMEOUT_MS
+);
+
+const pipeSseResponse = async (
+  response: Response,
+  signal: AbortSignal,
+  onChunk: (chunk: string) => void,
+  stallTimeoutMs?: number,
+): Promise<void> => {
   if (!response.body) {
     throw new Error('OpenCode SSE response missing body');
   }
 
   const reader = response.body.getReader();
   const decoder = new TextDecoder();
+  let stalled = false;
+  let stallTimer: ReturnType<typeof setTimeout> | null = null;
+
+  const clearStallTimer = () => {
+    if (!stallTimer) {
+      return;
+    }
+    clearTimeout(stallTimer);
+    stallTimer = null;
+  };
+
+  const resetStallTimer = () => {
+    clearStallTimer();
+    const timeoutMs = resolveStallTimeoutMs(stallTimeoutMs);
+    if (timeoutMs <= 0) {
+      return;
+    }
+    stallTimer = setTimeout(() => {
+      stalled = true;
+      void reader.cancel().catch(() => {});
+    }, timeoutMs);
+  };
 
   try {
+    resetStallTimer();
     while (!signal.aborted) {
       const { done, value } = await reader.read();
       if (done) {
         break;
       }
       if (value && value.length > 0) {
+        resetStallTimer();
         const chunk = decoder.decode(value, { stream: true });
         if (chunk.length > 0) {
           onChunk(chunk);
@@ -143,7 +178,12 @@ const pipeSseResponse = async (response: Response, signal: AbortSignal, onChunk:
     if (!signal.aborted && remaining.length > 0) {
       onChunk(remaining);
     }
+  } catch (error) {
+    if (!stalled) {
+      throw error;
+    }
   } finally {
+    clearStallTimer();
     try {
       await reader.cancel();
     } catch {
@@ -163,6 +203,7 @@ export const openSseProxy = async ({
   headers,
   signal,
   onChunk,
+  stallTimeoutMs,
 }: OpenSseProxyOptions): Promise<OpenSseProxyResult> => {
   // Reconnect logic with exponential backoff
   let reconnectAttempts = 0;
@@ -208,7 +249,7 @@ export const openSseProxy = async ({
   const run = (async () => {
     let activeResponse = response;
     try {
-      await pipeSseResponse(activeResponse, signal, onChunk);
+      await pipeSseResponse(activeResponse, signal, onChunk, stallTimeoutMs);
     } catch (error: unknown) {
       const cause = (error as { cause?: { code?: string } } | null)?.cause;
 
@@ -228,7 +269,7 @@ export const openSseProxy = async ({
             // Attempt to reconnect
             try {
               activeResponse = await connect();
-              await pipeSseResponse(activeResponse, signal, onChunk);
+              await pipeSseResponse(activeResponse, signal, onChunk, stallTimeoutMs);
               return; // Successfully reconnected
             } catch (reconnectError) {
               console.error('[SSE] Reconnect failed', reconnectError);