From 98c716cf9250b9d33f25eb707b74037c5be6b6c4 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Wed, 29 Jul 2026 18:38:05 +0300 Subject: [PATCH] fix(vscode): recover stalled OpenCode SSE streams --- packages/vscode/src/DOCUMENTATION.md | 1 + packages/vscode/src/sseProxy.test.ts | 101 +++++++++++++++++++++++++++ packages/vscode/src/sseProxy.ts | 47 ++++++++++++- 3 files changed, 146 insertions(+), 3 deletions(-) create mode 100644 packages/vscode/src/sseProxy.test.ts diff --git a/packages/vscode/src/DOCUMENTATION.md b/packages/vscode/src/DOCUMENTATION.md index 00de98d8..e8ff0ca2 100644 --- a/packages/vscode/src/DOCUMENTATION.md +++ b/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/*`). diff --git a/packages/vscode/src/sseProxy.test.ts b/packages/vscode/src/sseProxy.test.ts new file mode 100644 index 00000000..f26b1b12 --- /dev/null +++ b/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({}), { + 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({ + 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; + } + }); +}); diff --git a/packages/vscode/src/sseProxy.ts b/packages/vscode/src/sseProxy.ts index a4d66c30..009c730b 100644 --- a/packages/vscode/src/sseProxy.ts +++ b/packages/vscode/src/sseProxy.ts @@ -7,6 +7,7 @@ type OpenSseProxyOptions = { headers?: Record; 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((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 => { +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 => { 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 | 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 => { // 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);