From db2464d94d1a7046ebf2b9d2c2937c2861183107 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Tue, 26 May 2026 11:42:20 +0300 Subject: [PATCH] fix: restore live streaming in VS Code Forward SSE chunks without reserializing events Prevent generic proxy from buffering SSE endpoints Add regression tests for VS Code stream proxy --- .../vscode/src/bridge-proxy-runtime.test.js | 32 +++ packages/vscode/src/bridge-proxy-runtime.ts | 28 ++- packages/vscode/src/sseProxy.test.js | 85 ++++++++ packages/vscode/src/sseProxy.ts | 198 ++++++++++-------- 4 files changed, 241 insertions(+), 102 deletions(-) create mode 100644 packages/vscode/src/bridge-proxy-runtime.test.js create mode 100644 packages/vscode/src/sseProxy.test.js diff --git a/packages/vscode/src/bridge-proxy-runtime.test.js b/packages/vscode/src/bridge-proxy-runtime.test.js new file mode 100644 index 00000000..b56aa517 --- /dev/null +++ b/packages/vscode/src/bridge-proxy-runtime.test.js @@ -0,0 +1,32 @@ +import { describe, expect, it, mock } from 'bun:test'; + +const { handleProxyBridgeMessage } = await import('./bridge-proxy-runtime'); + +const createDeps = () => ({ + tryHandleLocalFsProxy: mock(() => Promise.resolve(null)), + buildUnavailableApiResponse: mock(() => ({ status: 503, headers: {}, bodyText: '' })), + sanitizeForwardHeaders: mock((headers) => headers || {}), + collectHeaders: mock(() => ({})), + base64EncodeUtf8: mock((text) => Buffer.from(text, 'utf8').toString('base64')), +}); + +describe('bridge proxy runtime', () => { + it('does not buffer SSE endpoints through the generic API proxy', async () => { + const deps = createDeps(); + + const response = await handleProxyBridgeMessage( + { id: '1', type: 'api:proxy', payload: { method: 'GET', path: '/global/event?lastEventId=evt-1' } }, + undefined, + deps, + ); + + expect(response?.success).toBe(true); + expect(response?.data).toMatchObject({ + status: 400, + headers: { 'content-type': 'application/json' }, + bodyText: JSON.stringify({ error: 'SSE requests must use api:sse:start' }), + }); + expect(deps.tryHandleLocalFsProxy).not.toHaveBeenCalled(); + expect(deps.buildUnavailableApiResponse).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/vscode/src/bridge-proxy-runtime.ts b/packages/vscode/src/bridge-proxy-runtime.ts index 65fb05cf..f40630ea 100644 --- a/packages/vscode/src/bridge-proxy-runtime.ts +++ b/packages/vscode/src/bridge-proxy-runtime.ts @@ -42,6 +42,15 @@ const collectProxyResponseHeaders = (headers: Headers, deps: Pick { + try { + const parsed = new URL(requestPath, 'https://openchamber.invalid'); + return parsed.pathname === '/event' || parsed.pathname === '/global/event'; + } catch { + return requestPath === '/event' || requestPath === '/global/event'; + } +}; + type ProxyRuntimeDeps = { tryHandleLocalFsProxy: (method: string, requestPath: string) => Promise; buildUnavailableApiResponse: () => ApiProxyResponsePayload; @@ -65,9 +74,18 @@ export async function handleProxyBridgeMessage( typeof requestPath === 'string' && requestPath.trim().length > 0 ? requestPath.trim().startsWith('/') ? requestPath.trim() - : `/${requestPath.trim()}` + : `/${requestPath.trim()}` : '/'; + if (isSseProxyPath(normalizedPath)) { + const data: ApiProxyResponsePayload = { + status: 400, + headers: { 'content-type': 'application/json' }, + bodyText: JSON.stringify({ error: 'SSE requests must use api:sse:start' }), + }; + return { id, type, success: true, data }; + } + const localFsResponse = await deps.tryHandleLocalFsProxy(normalizedMethod, normalizedPath); if (localFsResponse) { return { id, type, success: true, data: localFsResponse }; @@ -86,14 +104,6 @@ export async function handleProxyBridgeMessage( ...ctx?.manager?.getOpenCodeAuthHeaders(), }; - if (normalizedPath === '/event' || normalizedPath === '/global/event') { - if (!requestHeaders.Accept) { - requestHeaders.Accept = 'text/event-stream'; - } - requestHeaders['Cache-Control'] = requestHeaders['Cache-Control'] || 'no-cache'; - requestHeaders.Connection = requestHeaders.Connection || 'keep-alive'; - } - try { const response = await fetch(targetUrl, { method: normalizedMethod, diff --git a/packages/vscode/src/sseProxy.test.js b/packages/vscode/src/sseProxy.test.js new file mode 100644 index 00000000..e1d95767 --- /dev/null +++ b/packages/vscode/src/sseProxy.test.js @@ -0,0 +1,85 @@ +import { afterEach, describe, expect, it, mock } from 'bun:test'; + +const originalFetch = globalThis.fetch; +const { openSseProxy } = await import('./sseProxy'); + +const createManager = () => ({ + getApiUrl: () => 'http://127.0.0.1:4096/', + getWorkingDirectory: () => '/repo', + getOpenCodeAuthHeaders: () => ({ Authorization: 'Bearer test-token' }), + onStatusChange: () => ({ dispose() {} }), +}); + +const createSseResponse = (chunks) => { + const encoder = new TextEncoder(); + return new Response(new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)); + } + controller.close(); + }, + }), { + status: 200, + headers: { 'content-type': 'text/event-stream; charset=utf-8' }, + }); +}; + +describe('VS Code SSE proxy', () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + it('forwards upstream SSE chunks without reserializing event data', async () => { + const upstreamChunks = [ + 'id: evt-1\n', + 'data: {"type":"message.part.delta","properties":{"delta":"hi"}}\n\n', + ]; + let fetchInput; + let fetchInit; + globalThis.fetch = mock((input, init) => { + fetchInput = input; + fetchInit = init; + return Promise.resolve(createSseResponse(upstreamChunks)); + }); + + const received = []; + const controller = new AbortController(); + const proxy = await openSseProxy({ + manager: createManager(), + path: '/global/event', + headers: { 'Last-Event-ID': 'evt-0' }, + signal: controller.signal, + onChunk: (chunk) => received.push(chunk), + }); + + await proxy.run; + + expect(fetchInput).toBe('http://127.0.0.1:4096/global/event'); + expect(fetchInit.headers.Authorization).toBe('Bearer test-token'); + expect(fetchInit.headers['Last-Event-ID']).toBe('evt-0'); + expect(proxy.headers['content-type']).toContain('text/event-stream'); + expect(received.join('')).toBe(upstreamChunks.join('')); + }); + + it('adds the active directory for directory-scoped event streams', async () => { + let fetchInput; + globalThis.fetch = mock((input) => { + fetchInput = input; + return Promise.resolve(createSseResponse(['data: {"type":"server.connected"}\n\n'])); + }); + + const proxy = await openSseProxy({ + manager: createManager(), + path: '/event?foo=bar', + signal: new AbortController().signal, + onChunk: () => {}, + }); + await proxy.run; + + const url = new URL(fetchInput); + expect(url.pathname).toBe('/event'); + expect(url.searchParams.get('foo')).toBe('bar'); + expect(url.searchParams.get('directory')).toBe('/repo'); + }); +}); diff --git a/packages/vscode/src/sseProxy.ts b/packages/vscode/src/sseProxy.ts index ead1d4d1..a4d66c30 100644 --- a/packages/vscode/src/sseProxy.ts +++ b/packages/vscode/src/sseProxy.ts @@ -1,14 +1,6 @@ -import { createOpencodeClient } from '@opencode-ai/sdk/v2'; import type { OpenCodeManager } from './opencode'; import { waitForApiUrl } from './opencode-ready'; -type StreamEvent = { - data: TData; - event?: string; - id?: string; - retry?: number; -}; - type OpenSseProxyOptions = { manager: OpenCodeManager; path: string; @@ -50,27 +42,13 @@ const sleep = (ms: number, signal: AbortSignal) => new Promise((resolve) = const getAbortReason = (signal: AbortSignal) => signal.reason ?? new DOMException('Aborted', 'AbortError'); -const serializeSseEventBlock = (event: StreamEvent): string => { - const lines: string[] = []; - if (typeof event.id === 'string' && event.id.length > 0) { - lines.push(`id: ${event.id}`); - } - if (typeof event.event === 'string' && event.event.length > 0) { - lines.push(`event: ${event.event}`); - } - if (typeof event.retry === 'number' && Number.isFinite(event.retry)) { - lines.push(`retry: ${event.retry}`); - } - lines.push(`data: ${JSON.stringify(event.data)}`); - return lines.join('\n'); -}; - -const normalizeSsePath = (path: string): { pathname: '/event' | '/global/event'; directory: string | null } => { +const normalizeSsePath = (path: string): { pathname: '/event' | '/global/event'; searchParams: URLSearchParams; directory: string | null } => { const parsed = new URL(path, 'https://openchamber.invalid'); const pathname = parsed.pathname === '/global/event' ? '/global/event' : '/event'; const directory = parsed.searchParams.get('directory'); return { pathname, + searchParams: new URLSearchParams(parsed.searchParams), directory: typeof directory === 'string' && directory.trim().length > 0 ? directory.trim() : null, }; }; @@ -79,41 +57,105 @@ const resolveDefaultDirectory = (manager: OpenCodeManager): string => { return manager.getWorkingDirectory() || 'global'; }; -const createAuthedClient = async (manager: OpenCodeManager, headers?: Record) => { +const createSseUrl = (baseUrl: string, pathname: '/event' | '/global/event', searchParams: URLSearchParams, directory: string): URL => { + const base = `${baseUrl.replace(/\/+$/, '')}/`; + const url = new URL(pathname.replace(/^\/+/, ''), base); + for (const [key, value] of searchParams) { + url.searchParams.append(key, value); + } + if (pathname === '/event' && !url.searchParams.has('directory')) { + url.searchParams.set('directory', directory); + } + return url; +}; + +const createSseHeaders = (manager: OpenCodeManager, headers?: Record): Record => ({ + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + ...(headers || {}), + ...manager.getOpenCodeAuthHeaders(), +}); + +const createSseResponseHeaders = (response: Response): Record => ({ + 'content-type': response.headers.get('content-type') || SSE_RESPONSE_HEADERS['content-type'], + 'cache-control': response.headers.get('cache-control') || SSE_RESPONSE_HEADERS['cache-control'], +}); + +const fetchSseResponse = async ( + manager: OpenCodeManager, + path: string, + headers: Record | undefined, + signal: AbortSignal, +): Promise => { const baseUrl = await waitForApiUrl(manager); if (!baseUrl) { throw new Error('OpenCode API URL not available'); } - return createOpencodeClient({ - baseUrl, - headers: { - ...(headers || {}), - ...manager.getOpenCodeAuthHeaders(), - }, + const { pathname, searchParams, directory } = normalizeSsePath(path); + const resolvedDirectory = directory || resolveDefaultDirectory(manager); + const targetUrl = createSseUrl(baseUrl, pathname, searchParams, resolvedDirectory); + + const response = await fetch(targetUrl.toString(), { + method: 'GET', + headers: createSseHeaders(manager, headers), + signal, }); + + if (!response.ok) { + await response.body?.cancel().catch(() => {}); + const error = new Error(`OpenCode SSE request failed (${response.status})`); + (error as Error & { status?: number }).status = response.status; + throw error; + } + + if (!response.body) { + throw new Error('OpenCode SSE response missing body'); + } + + return response; }; -const getSseOptions = ( - signal: AbortSignal, - onChunk: (chunk: string) => void, - wrapDirectory?: string, -) => ({ - signal, - sseMaxRetryAttempts: 0, - onSseEvent: (event: StreamEvent) => { - const nextEvent = wrapDirectory - ? { - ...event, - data: { - directory: wrapDirectory, - payload: event.data, - }, +const pipeSseResponse = async (response: Response, signal: AbortSignal, onChunk: (chunk: string) => void): Promise => { + if (!response.body) { + throw new Error('OpenCode SSE response missing body'); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + + try { + while (!signal.aborted) { + const { done, value } = await reader.read(); + if (done) { + break; + } + if (value && value.length > 0) { + const chunk = decoder.decode(value, { stream: true }); + if (chunk.length > 0) { + onChunk(chunk); } - : event; - onChunk(`${serializeSseEventBlock(nextEvent)}\n\n`); - }, -}); + } + } + + const remaining = decoder.decode(); + if (!signal.aborted && remaining.length > 0) { + onChunk(remaining); + } + } finally { + try { + await reader.cancel(); + } catch { + // ignore cancel failures during stream shutdown + } + try { + reader.releaseLock(); + } catch { + // ignore release failures after reader shutdown + } + } +}; export const openSseProxy = async ({ manager, @@ -122,45 +164,22 @@ export const openSseProxy = async ({ signal, onChunk, }: OpenSseProxyOptions): Promise => { - const client = await createAuthedClient(manager, headers); - const { pathname, directory } = normalizeSsePath(path); - const resolvedDirectory = directory || resolveDefaultDirectory(manager); - // Reconnect logic with exponential backoff let reconnectAttempts = 0; - const connect = async (): Promise<{ stream: AsyncIterable }> => { + const connect = async (): Promise => { try { + const { pathname } = normalizeSsePath(path); console.log(`[SSE] Connecting to ${pathname} (attempt ${reconnectAttempts + 1}/${MAX_RECONNECTS + 1})`); - if (pathname === '/global/event') { - try { - const result = await client.global.event(getSseOptions(signal, onChunk)); - // Reset reconnect counter on successful connection - reconnectAttempts = 0; - return result; - } catch (error) { - if ((error as Error)?.name === 'AbortError' || signal.aborted) { - throw error; - } - // Fallback to directory event on error - console.warn('[SSE] Global event failed, falling back to directory event', error); - const result = client.event.subscribe( - { directory: resolvedDirectory }, - getSseOptions(signal, onChunk, resolvedDirectory), - ); - reconnectAttempts = 0; - return result; - } - } - - const result = client.event.subscribe( - { directory: resolvedDirectory }, - getSseOptions(signal, onChunk), - ); + const result = await fetchSseResponse(manager, path, headers, signal); reconnectAttempts = 0; return result; } catch (error) { + if ((error as Error)?.name === 'AbortError' || signal.aborted) { + throw error; + } + // Implement reconnect logic if (!signal.aborted && reconnectAttempts < MAX_RECONNECTS) { reconnectAttempts++; @@ -184,16 +203,12 @@ export const openSseProxy = async ({ } }; - const result = await connect(); + const response = await connect(); const run = (async () => { + let activeResponse = response; try { - for await (const _ of result.stream) { - void _; - if (signal.aborted) { - break; - } - } + await pipeSseResponse(activeResponse, signal, onChunk); } catch (error: unknown) { const cause = (error as { cause?: { code?: string } } | null)?.cause; @@ -212,11 +227,8 @@ export const openSseProxy = async ({ // Attempt to reconnect try { - const newResult = await connect(); - for await (const _ of newResult.stream) { - void _; - if (signal.aborted) break; - } + activeResponse = await connect(); + await pipeSseResponse(activeResponse, signal, onChunk); return; // Successfully reconnected } catch (reconnectError) { console.error('[SSE] Reconnect failed', reconnectError); @@ -231,7 +243,7 @@ export const openSseProxy = async ({ })(); return { - headers: { ...SSE_RESPONSE_HEADERS }, + headers: createSseResponseHeaders(response), run, }; };