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
2026-05-26 11:42:20 +03:00
parent c61f16e0dd
commit db2464d94d
4 changed files with 241 additions and 102 deletions
@@ -1,14 +1,6 @@
-import { createOpencodeClient } from '@opencode-ai/sdk/v2';
 import type { OpenCodeManager } from './opencode';
 import { waitForApiUrl } from './opencode-ready';

-type StreamEvent<TData = unknown> = {
-  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<void>((resolve) =

 const getAbortReason = (signal: AbortSignal) => signal.reason ?? new DOMException('Aborted', 'AbortError');

-const serializeSseEventBlock = (event: StreamEvent<unknown>): 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<string, string>) => {
+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<string, string>): Record<string, string> => ({
+  Accept: 'text/event-stream',
+  'Cache-Control': 'no-cache',
+  Connection: 'keep-alive',
+  ...(headers || {}),
+  ...manager.getOpenCodeAuthHeaders(),
+});
+
+const createSseResponseHeaders = (response: Response): Record<string, string> => ({
+  '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<string, string> | undefined,
+  signal: AbortSignal,
+): Promise<Response> => {
  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<unknown>) => {
-    const nextEvent = wrapDirectory
-      ? {
-          ...event,
-          data: {
-            directory: wrapDirectory,
-            payload: event.data,
-          },
+const pipeSseResponse = async (response: Response, signal: AbortSignal, onChunk: (chunk: string) => void): Promise<void> => {
+  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<OpenSseProxyResult> => {
-  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<unknown> }> => {
+  const connect = async (): Promise<Response> => {
    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,
  };
 };