From 7030549d789ffa116ae950fe93c4083b19fa4a6c Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Fri, 24 Apr 2026 11:58:16 +0300 Subject: [PATCH] Share upstream event stream hub --- packages/web/server/index.js | 8 + .../server/lib/event-stream/DOCUMENTATION.md | 21 +- .../web/server/lib/event-stream/global-hub.js | 143 ++++++ packages/web/server/lib/event-stream/index.js | 11 + .../web/server/lib/event-stream/runtime.js | 422 +++++++++++------- .../server/lib/event-stream/runtime.test.js | 317 +++++++++++++ .../lib/event-stream/upstream-reader.js | 204 +++++++++ .../lib/event-stream/upstream-reader.test.js | 159 +++++++ .../web/server/lib/opencode/DOCUMENTATION.md | 8 +- .../lib/opencode/startup-pipeline-runtime.js | 2 + packages/web/server/lib/opencode/watcher.js | 103 +++-- .../web/server/lib/opencode/watcher.test.js | 239 ++++++++++ 12 files changed, 1436 insertions(+), 201 deletions(-) create mode 100644 packages/web/server/lib/event-stream/global-hub.js create mode 100644 packages/web/server/lib/event-stream/upstream-reader.js create mode 100644 packages/web/server/lib/event-stream/upstream-reader.test.js create mode 100644 packages/web/server/lib/opencode/watcher.test.js diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 0751983b..1ba86b41 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -33,6 +33,7 @@ import { detectSayTtsCapability } from './lib/tts/capability-runtime.js'; import { createTerminalRuntime } from './lib/terminal/runtime.js'; import { createGlobalUiEventBroadcaster, + createGlobalMessageStreamHub, createMessageStreamWsRuntime, } from './lib/event-stream/index.js'; import { createFsSearchRuntime as createFsSearchRuntimeFactory } from './lib/fs/search.js'; @@ -648,11 +649,17 @@ const notificationTriggerRuntime = createNotificationTriggerRuntime({ const maybeSendPushForTrigger = (...args) => notificationTriggerRuntime.maybeSendPushForTrigger(...args); const setAutoAcceptSession = (...args) => notificationTriggerRuntime.setAutoAcceptSession(...args); +const globalMessageStreamHub = createGlobalMessageStreamHub({ + buildOpenCodeUrl, + getOpenCodeAuthHeaders, +}); + const openCodeWatcherRuntime = createOpenCodeWatcherRuntime({ waitForOpenCodePort: (...args) => waitForOpenCodePort(...args), buildOpenCodeUrl, getOpenCodeAuthHeaders, parseSseDataPayload: (...args) => parseSseDataPayload(...args), + globalEventHub: globalMessageStreamHub, onPayload: (payload) => { maybeCacheSessionInfoFromEvent(payload); void maybeSendPushForTrigger(payload); @@ -1155,6 +1162,7 @@ async function main(options = {}) { rejectWebSocketUpgrade, buildOpenCodeUrl, getOpenCodeAuthHeaders, + globalEventHub: globalMessageStreamHub, processForwardedEventPayload, messageStreamWsClients: uiNotificationWsClients, terminalHeartbeatIntervalMs: TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS, diff --git a/packages/web/server/lib/event-stream/DOCUMENTATION.md b/packages/web/server/lib/event-stream/DOCUMENTATION.md index e95a2099..921d9c9b 100644 --- a/packages/web/server/lib/event-stream/DOCUMENTATION.md +++ b/packages/web/server/lib/event-stream/DOCUMENTATION.md @@ -5,9 +5,12 @@ This module contains the OpenChamber message-stream WebSocket protocol and runti ## Entrypoints and structure - `packages/web/server/lib/event-stream/index.js`: public entrypoint re-exporting protocol and runtime helpers. +- `packages/web/server/lib/event-stream/global-hub.js`: shared global upstream SSE hub for server-side subscribers and browser WS fan-out. - `packages/web/server/lib/event-stream/protocol.js`: path constants, SSE envelope parsing, and WebSocket frame serialization helpers. -- `packages/web/server/lib/event-stream/runtime.js`: WebSocket server runtime, upgrade handling, SSE-to-WS bridging, and global event broadcasting. +- `packages/web/server/lib/event-stream/upstream-reader.js`: reusable upstream SSE reader with event-id tracking, stall recovery, and reconnect handling. +- `packages/web/server/lib/event-stream/runtime.js`: WebSocket server runtime, upgrade handling, shared global upstream hub, per-directory upstream reader orchestration, and global event broadcasting. - `packages/web/server/lib/event-stream/protocol.test.js`: unit tests for protocol helpers. +- `packages/web/server/lib/event-stream/upstream-reader.test.js`: unit tests for upstream SSE reader behavior. - `packages/web/server/lib/event-stream/runtime.test.js`: unit tests for runtime-side broadcaster behavior. ## Public exports @@ -21,25 +24,35 @@ This module contains the OpenChamber message-stream WebSocket protocol and runti - `sendMessageStreamWsEvent(socket, payload, options)`: sends an event frame with optional `eventId` and `directory`. ### Runtime helpers +- `createGlobalMessageStreamHub(...)`: creates a shared `/global/event` upstream SSE hub with event/status subscribers and bounded event-id replay. - `createGlobalUiEventBroadcaster({ sseClients, wsClients, writeSseEvent })`: returns a broadcaster that fans out the same synthetic UI event to SSE and WS clients. - `createMessageStreamWsRuntime(...)`: mounts the message-stream WS server, upgrade handler, and SSE-to-WS bridge onto the web HTTP server. +### Upstream reader helpers +- `DEFAULT_UPSTREAM_STALL_TIMEOUT_MS`: default idle timeout before an attached upstream SSE fetch is aborted for reconnect. +- `DEFAULT_UPSTREAM_RECONNECT_DELAY_MS`: default delay between upstream reconnect attempts. +- `createUpstreamSseReader(...)`: creates a start/stop reader for OpenCode SSE streams. The reader parses SSE blocks, tracks the latest `Last-Event-ID`, reconnects after closed or stalled upstream streams, and reports events through callbacks. + ## Runtime behavior - Browser clients connect to the WS endpoints above. - OpenChamber still fetches OpenCode upstream event streams over SSE. -- Each WS connection proxies one upstream SSE stream. -- If an upstream SSE stream stalls after the browser WS is already ready, the runtime aborts that upstream fetch and reconnects upstream with `Last-Event-ID`, keeping the browser WS alive when recovery is fast. +- The web server creates one shared global message-stream hub. OpenCode watcher side effects and global WS clients subscribe to that hub, so there is one upstream `/global/event` SSE reader for both server-side processing and browser fan-out. +- The global hub keeps a bounded replay buffer keyed by SSE `eventId` so reconnecting browser clients can receive buffered events after their requested `Last-Event-ID`. +- Directory WS clients still attach one upstream `/event?directory=...` SSE reader per connection because directory streams are scoped. +- If an upstream SSE stream stalls after the browser WS is already ready, the reader aborts that upstream fetch and reconnects upstream with `Last-Event-ID`, keeping the browser WS alive when recovery is fast. - Health checks are reserved for initial upstream connect failures and explicit upstream-unavailable responses, not for ordinary stall recovery on an already-established stream. - Global synthetic events such as `openchamber:session-status`, `openchamber:session-activity`, `openchamber:notification`, and `openchamber:heartbeat` are preserved on the WS path, but heartbeat frames are emitted only while an upstream SSE stream is actively attached. - Global UI broadcasts are fan-out capable across both SSE and WS clients. +- The reusable upstream reader centralizes SSE fetch/parsing/reconnect behavior for the WS runtime and OpenCode watcher. Additional event consumers should move to it only with parity tests for their lifecycle and error semantics. ## Notes for contributors - Keep protocol helpers pure and small so they can be unit tested without spinning up a server. - Keep runtime wiring in this module instead of `packages/web/server/index.js` unless the logic is strictly route-local glue. - Do not change upstream OpenCode transport assumptions here; OpenCode remains SSE-based. -- If replay support is added later, add it here rather than growing `index.js`. +- Keep global replay bounded; do not turn it into an unbounded event log. ## Testing - Run `bun test packages/web/server/lib/event-stream/protocol.test.js` +- Run `bun test packages/web/server/lib/event-stream/upstream-reader.test.js` - Run `bun test packages/web/server/lib/event-stream/runtime.test.js` - Run repo validation before finalizing: `bun run type-check`, `bun run lint`, `bun run build` diff --git a/packages/web/server/lib/event-stream/global-hub.js b/packages/web/server/lib/event-stream/global-hub.js new file mode 100644 index 00000000..1bc98abe --- /dev/null +++ b/packages/web/server/lib/event-stream/global-hub.js @@ -0,0 +1,143 @@ +import { createUpstreamSseReader } from './upstream-reader.js'; + +export const MESSAGE_STREAM_GLOBAL_REPLAY_LIMIT = 512; + +export function createGlobalMessageStreamHub({ + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + fetchImpl = fetch, + upstreamStallTimeoutMs, + upstreamReconnectDelayMs, + replayLimit = MESSAGE_STREAM_GLOBAL_REPLAY_LIMIT, +}) { + const eventSubscribers = new Set(); + const statusSubscribers = new Set(); + const replay = []; + + let controller = null; + let reader = null; + let connected = false; + let everConnected = false; + let buildUrlFailed = false; + + const notifyStatus = (status) => { + for (const subscriber of Array.from(statusSubscribers)) { + subscriber(status); + } + }; + + const normalizeEvent = ({ envelope, payload }) => { + const directory = + typeof envelope?.directory === 'string' && envelope.directory.length > 0 ? envelope.directory : 'global'; + const eventId = typeof envelope?.eventId === 'string' && envelope.eventId.length > 0 ? envelope.eventId : undefined; + return { + envelope, + payload, + directory, + eventId, + }; + }; + + const start = () => { + if (reader) { + return; + } + + controller = new AbortController(); + reader = createUpstreamSseReader({ + signal: controller.signal, + stallTimeoutMs: upstreamStallTimeoutMs, + reconnectDelayMs: upstreamReconnectDelayMs, + fetchImpl, + buildUrl: () => { + buildUrlFailed = false; + try { + return new URL(buildOpenCodeUrl('/global/event', '')); + } catch { + buildUrlFailed = true; + throw new Error('OpenCode service unavailable'); + } + }, + getHeaders: getOpenCodeAuthHeaders, + onConnect() { + connected = true; + const wasReady = everConnected; + everConnected = true; + notifyStatus({ type: 'connect', wasReady }); + }, + onDisconnect({ reason }) { + connected = false; + notifyStatus({ type: 'disconnect', reason }); + }, + onEvent(event) { + const normalized = normalizeEvent(event); + if (normalized.eventId) { + replay.push(normalized); + if (replay.length > replayLimit) { + replay.splice(0, replay.length - replayLimit); + } + } + + for (const subscriber of Array.from(eventSubscribers)) { + subscriber(normalized); + } + }, + onError(error) { + if (controller?.signal.aborted) { + return; + } + + notifyStatus({ + type: everConnected ? 'error' : 'initial-error', + error, + buildUrlFailed, + }); + }, + }); + + void reader.start(); + }; + + const stop = () => { + connected = false; + reader?.stop(); + if (controller && !controller.signal.aborted) { + controller.abort(); + } + reader = null; + controller = null; + everConnected = false; + buildUrlFailed = false; + }; + + return { + start, + stop, + isConnected() { + return connected; + }, + hasConnected() { + return everConnected; + }, + subscribeEvent(subscriber) { + eventSubscribers.add(subscriber); + return () => { + eventSubscribers.delete(subscriber); + }; + }, + subscribeStatus(subscriber) { + statusSubscribers.add(subscriber); + return () => { + statusSubscribers.delete(subscriber); + }; + }, + replayAfter(eventId) { + if (!eventId) { + return []; + } + + const index = replay.findIndex((entry) => entry.eventId === eventId); + return index === -1 ? [] : replay.slice(index + 1); + }, + }; +} diff --git a/packages/web/server/lib/event-stream/index.js b/packages/web/server/lib/event-stream/index.js index ec6e3fa3..e55309f0 100644 --- a/packages/web/server/lib/event-stream/index.js +++ b/packages/web/server/lib/event-stream/index.js @@ -11,3 +11,14 @@ export { createGlobalUiEventBroadcaster, createMessageStreamWsRuntime, } from './runtime.js'; + +export { + MESSAGE_STREAM_GLOBAL_REPLAY_LIMIT, + createGlobalMessageStreamHub, +} from './global-hub.js'; + +export { + DEFAULT_UPSTREAM_RECONNECT_DELAY_MS, + DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + createUpstreamSseReader, +} from './upstream-reader.js'; diff --git a/packages/web/server/lib/event-stream/runtime.js b/packages/web/server/lib/event-stream/runtime.js index cee8f6f1..21df5d26 100644 --- a/packages/web/server/lib/event-stream/runtime.js +++ b/packages/web/server/lib/event-stream/runtime.js @@ -5,13 +5,15 @@ import { MESSAGE_STREAM_DIRECTORY_WS_PATH, MESSAGE_STREAM_GLOBAL_WS_PATH, MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS, - parseSseEventEnvelope, sendMessageStreamWsEvent, sendMessageStreamWsFrame, } from './protocol.js'; - -const MESSAGE_STREAM_UPSTREAM_STALL_TIMEOUT_MS = 20_000; -const MESSAGE_STREAM_UPSTREAM_RECONNECT_DELAY_MS = 250; +import { createGlobalMessageStreamHub } from './global-hub.js'; +import { + DEFAULT_UPSTREAM_RECONNECT_DELAY_MS, + DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + createUpstreamSseReader, +} from './upstream-reader.js'; function shouldTriggerUpstreamHealthCheck(upstream) { if (!upstream) { @@ -71,14 +73,161 @@ export function createMessageStreamWsRuntime({ wsClients, triggerHealthCheck, heartbeatIntervalMs = MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS, - upstreamStallTimeoutMs = MESSAGE_STREAM_UPSTREAM_STALL_TIMEOUT_MS, - upstreamReconnectDelayMs = MESSAGE_STREAM_UPSTREAM_RECONNECT_DELAY_MS, + upstreamStallTimeoutMs = DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + upstreamReconnectDelayMs = DEFAULT_UPSTREAM_RECONNECT_DELAY_MS, fetchImpl = fetch, + globalEventHub = null, }) { const wsServer = new WebSocketServer({ noServer: true, }); + const ownsGlobalHub = !globalEventHub; + const globalHub = globalEventHub ?? createGlobalMessageStreamHub({ + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + fetchImpl, + upstreamStallTimeoutMs, + upstreamReconnectDelayMs, + }); + + const globalClients = new Set(); + const globalClientLastEventIds = new Map(); + const globalReadyClients = new Set(); + + const replayGlobalEvents = (socket, requestedLastEventId) => { + for (const entry of globalHub.replayAfter(requestedLastEventId)) { + const sent = sendMessageStreamWsEvent(socket, entry.payload, { + directory: entry.directory, + eventId: entry.eventId, + }); + if (!sent) { + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + return; + } + } + }; + + const markGlobalClientReady = (socket, requestedLastEventId) => { + if (socket.readyState !== 1) { + return; + } + + const sent = sendMessageStreamWsFrame(socket, { + type: 'ready', + scope: 'global', + }); + if (!sent) { + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + return; + } + + globalReadyClients.add(socket); + wsClients.add(socket); + replayGlobalEvents(socket, requestedLastEventId); + }; + + const closeGlobalClientsWithInitialError = ({ message, closeReason = message, triggerHealthCheckFor = null }) => { + for (const socket of Array.from(globalClients)) { + sendMessageStreamWsFrame(socket, { type: 'error', message }); + try { + socket.close(1011, closeReason); + } catch { + } + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + } + + if (triggerHealthCheckFor === true || (triggerHealthCheckFor && shouldTriggerUpstreamHealthCheck(triggerHealthCheckFor))) { + triggerHealthCheck?.(); + } + + if (ownsGlobalHub) { + globalHub.stop(); + } + }; + + const unsubscribeGlobalEvent = globalHub.subscribeEvent(({ envelope, payload, directory, eventId }) => { + for (const socket of Array.from(globalClients)) { + if (!globalReadyClients.has(socket)) { + continue; + } + const sent = sendMessageStreamWsEvent(socket, payload, { + directory, + eventId, + }); + if (!sent) { + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + } + } + + processForwardedEventPayload(payload, (syntheticPayload) => { + for (const socket of Array.from(globalClients)) { + if (!globalReadyClients.has(socket)) { + continue; + } + const sent = sendMessageStreamWsEvent(socket, syntheticPayload, { directory: 'global' }); + if (!sent) { + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + } + } + }); + }); + + const unsubscribeGlobalStatus = globalHub.subscribeStatus((status) => { + if (status.type === 'connect') { + for (const socket of Array.from(globalClients)) { + if (!globalReadyClients.has(socket)) { + markGlobalClientReady(socket, globalClientLastEventIds.get(socket) ?? ''); + } + } + return; + } + + if (status.type === 'initial-error') { + const error = status.error; + if (error?.type === 'upstream_unavailable') { + closeGlobalClientsWithInitialError({ + message: `OpenCode event stream unavailable (${error.status})`, + closeReason: 'OpenCode event stream unavailable', + triggerHealthCheckFor: error.response, + }); + return; + } + + closeGlobalClientsWithInitialError({ + message: status.buildUrlFailed ? 'OpenCode service unavailable' : 'Failed to connect to OpenCode event stream', + closeReason: status.buildUrlFailed ? 'OpenCode service unavailable' : 'Failed to connect to OpenCode event stream', + triggerHealthCheckFor: !status.buildUrlFailed, + }); + return; + } + + if (status.type === 'error' && status.error?.type === 'stream_error') { + console.warn('Message stream WS proxy error:', status.error.error); + } + }); + + const stopGlobalHubIfUnused = () => { + if (ownsGlobalHub && globalClients.size === 0) { + globalHub.stop(); + } + }; + wsServer.on('connection', (socket, req) => { const rawUrl = typeof req?.url === 'string' ? req.url : MESSAGE_STREAM_GLOBAL_WS_PATH; const pathname = parseRequestPathname(rawUrl); @@ -87,23 +236,61 @@ export function createMessageStreamWsRuntime({ const requestedLastEventId = requestUrl.searchParams.get('lastEventId')?.trim() || ''; const requestedDirectory = requestUrl.searchParams.get('directory')?.trim() || ''; + if (isGlobalStream) { + const pingInterval = setInterval(() => { + if (socket.readyState !== 1) { + return; + } + + try { + socket.ping(); + } catch { + } + }, heartbeatIntervalMs); + + const heartbeatInterval = setInterval(() => { + if (!globalHub.isConnected()) { + return; + } + + sendMessageStreamWsEvent(socket, { type: 'openchamber:heartbeat', timestamp: Date.now() }, { directory: 'global' }); + }, heartbeatIntervalMs); + + socket.on('close', () => { + clearInterval(pingInterval); + clearInterval(heartbeatInterval); + globalClients.delete(socket); + globalClientLastEventIds.delete(socket); + globalReadyClients.delete(socket); + wsClients.delete(socket); + stopGlobalHubIfUnused(); + }); + + socket.on('error', () => { + void 0; + }); + + globalClients.add(socket); + globalClientLastEventIds.set(socket, requestedLastEventId); + globalHub.start(); + if (globalHub.isConnected()) { + markGlobalClientReady(socket, requestedLastEventId); + } + return; + } + const controller = new AbortController(); - let currentUpstreamAbortController = null; let upstreamConnected = false; let streamReady = false; - let lastEventId = requestedLastEventId; + let reader = null; const cleanup = () => { if (!controller.signal.aborted) { controller.abort(); } - if (currentUpstreamAbortController && !currentUpstreamAbortController.signal.aborted) { - currentUpstreamAbortController.abort(); - } + reader?.stop(); wsClients.delete(socket); }; - const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); - const pingInterval = setInterval(() => { if (socket.readyState !== 1) { return; @@ -135,25 +322,11 @@ export function createMessageStreamWsRuntime({ }); const run = async () => { - const forwardBlock = (block) => { - if (!block) { - return; - } - - const envelope = parseSseEventEnvelope(block); - const payload = envelope?.payload ?? null; - if (!payload) { - return; - } - + const forwardEvent = ({ envelope, payload }) => { const directory = isGlobalStream ? (typeof envelope?.directory === 'string' && envelope.directory.length > 0 ? envelope.directory : 'global') : (requestedDirectory || envelope?.directory || 'global'); - if (typeof envelope?.eventId === 'string' && envelope.eventId.length > 0) { - lastEventId = envelope.eventId; - } - sendMessageStreamWsEvent(socket, payload, { directory, eventId: typeof envelope?.eventId === 'string' && envelope.eventId.length > 0 ? envelope.eventId : undefined, @@ -165,149 +338,85 @@ export function createMessageStreamWsRuntime({ }; try { - while (!controller.signal.aborted) { - let targetUrl; - try { - targetUrl = new URL(buildOpenCodeUrl(isGlobalStream ? '/global/event' : '/event', '')); - } catch { - sendMessageStreamWsFrame(socket, { type: 'error', message: 'OpenCode service unavailable' }); - socket.close(1011, 'OpenCode service unavailable'); - return; + let buildUrlFailed = false; + const closeWithInitialError = ({ message, closeReason = message, triggerHealthCheckFor = null }) => { + sendMessageStreamWsFrame(socket, { type: 'error', message }); + socket.close(1011, closeReason); + if (triggerHealthCheckFor === true || (triggerHealthCheckFor && shouldTriggerUpstreamHealthCheck(triggerHealthCheckFor))) { + triggerHealthCheck?.(); } + reader?.stop(); + cleanup(); + }; - if (!isGlobalStream && requestedDirectory) { - targetUrl.searchParams.set('directory', requestedDirectory); - } + reader = createUpstreamSseReader({ + initialLastEventId: requestedLastEventId, + signal: controller.signal, + stallTimeoutMs: upstreamStallTimeoutMs, + reconnectDelayMs: upstreamReconnectDelayMs, + fetchImpl, + buildUrl: () => { + buildUrlFailed = false; + let targetUrl; + try { + targetUrl = new URL(buildOpenCodeUrl('/event', '')); + } catch { + buildUrlFailed = true; + throw new Error('OpenCode service unavailable'); + } - const headers = { - Accept: 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - ...getOpenCodeAuthHeaders(), - }; - if (lastEventId) { - headers['Last-Event-ID'] = lastEventId; - } + if (requestedDirectory) { + targetUrl.searchParams.set('directory', requestedDirectory); + } - const upstreamController = new AbortController(); - currentUpstreamAbortController = upstreamController; - const abortUpstream = () => upstreamController.abort(); - controller.signal.addEventListener('abort', abortUpstream, { once: true }); + return targetUrl; + }, + getHeaders: getOpenCodeAuthHeaders, + onConnect() { + if (!streamReady) { + sendMessageStreamWsFrame(socket, { + type: 'ready', + scope: 'directory', + }); + streamReady = true; + } - let upstream; - try { - upstream = await fetchImpl(targetUrl.toString(), { - headers, - signal: upstreamController.signal, - }); - } catch { - controller.signal.removeEventListener('abort', abortUpstream); - currentUpstreamAbortController = null; + upstreamConnected = true; + }, + onDisconnect() { upstreamConnected = false; + }, + onEvent: forwardEvent, + onError(error) { if (controller.signal.aborted) { return; } - if (!streamReady) { - sendMessageStreamWsFrame(socket, { type: 'error', message: 'Failed to connect to OpenCode event stream' }); - socket.close(1011, 'Failed to connect to OpenCode event stream'); - triggerHealthCheck?.(); - return; - } - await wait(upstreamReconnectDelayMs); - continue; - } - if (!upstream.ok || !upstream.body) { - controller.signal.removeEventListener('abort', abortUpstream); - currentUpstreamAbortController = null; - upstreamConnected = false; if (!streamReady) { - sendMessageStreamWsFrame(socket, { - type: 'error', - message: `OpenCode event stream unavailable (${upstream.status})`, + if (error?.type === 'upstream_unavailable') { + closeWithInitialError({ + message: `OpenCode event stream unavailable (${error.status})`, + closeReason: 'OpenCode event stream unavailable', + triggerHealthCheckFor: error.response, + }); + return; + } + + closeWithInitialError({ + message: buildUrlFailed ? 'OpenCode service unavailable' : 'Failed to connect to OpenCode event stream', + closeReason: buildUrlFailed ? 'OpenCode service unavailable' : 'Failed to connect to OpenCode event stream', + triggerHealthCheckFor: !buildUrlFailed, }); - socket.close(1011, 'OpenCode event stream unavailable'); - if (shouldTriggerUpstreamHealthCheck(upstream)) { - triggerHealthCheck?.(); - } return; } - await wait(upstreamReconnectDelayMs); - continue; - } - if (!streamReady) { - sendMessageStreamWsFrame(socket, { - type: 'ready', - scope: isGlobalStream ? 'global' : 'directory', - }); - streamReady = true; - - if (isGlobalStream) { - wsClients.add(socket); + if (error?.type === 'stream_error') { + console.warn('Message stream WS proxy error:', error.error); } - } + }, + }); - upstreamConnected = true; - const decoder = new TextDecoder(); - const reader = upstream.body.getReader(); - let buffer = ''; - let upstreamAbortReason = null; - let stallTimer = null; - const resetStallTimer = () => { - if (stallTimer) { - clearTimeout(stallTimer); - } - stallTimer = setTimeout(() => { - upstreamAbortReason = 'upstream_stalled'; - upstreamConnected = false; - upstreamController.abort(); - }, upstreamStallTimeoutMs); - }; - - resetStallTimer(); - - try { - while (true) { - const { value, done } = await reader.read(); - if (done) { - break; - } - - resetStallTimer(); - buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n'); - - let separatorIndex = buffer.indexOf('\n\n'); - while (separatorIndex !== -1) { - const block = buffer.slice(0, separatorIndex); - buffer = buffer.slice(separatorIndex + 2); - forwardBlock(block); - separatorIndex = buffer.indexOf('\n\n'); - } - } - - if (buffer.trim().length > 0) { - forwardBlock(buffer.trim()); - } - } catch (error) { - if (!controller.signal.aborted && upstreamAbortReason !== 'upstream_stalled') { - console.warn('Message stream WS proxy error:', error); - } - } finally { - if (stallTimer) { - clearTimeout(stallTimer); - } - upstreamConnected = false; - currentUpstreamAbortController = null; - controller.signal.removeEventListener('abort', abortUpstream); - } - - if (controller.signal.aborted) { - return; - } - - await wait(upstreamReconnectDelayMs); - } + await reader.start(); } catch (error) { if (!controller.signal.aborted) { console.warn('Message stream WS proxy error:', error); @@ -367,6 +476,11 @@ export function createMessageStreamWsRuntime({ wsServer, async close() { server.off('upgrade', upgradeHandler); + unsubscribeGlobalEvent(); + unsubscribeGlobalStatus(); + if (ownsGlobalHub) { + globalHub.stop(); + } try { for (const client of wsServer.clients) { diff --git a/packages/web/server/lib/event-stream/runtime.test.js b/packages/web/server/lib/event-stream/runtime.test.js index 5d3c366d..19065a92 100644 --- a/packages/web/server/lib/event-stream/runtime.test.js +++ b/packages/web/server/lib/event-stream/runtime.test.js @@ -127,6 +127,273 @@ describe('event stream broadcaster', () => { }); describe('message stream websocket runtime', () => { + it('shares one global upstream SSE reader across multiple websocket clients', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let fetchCalls = 0; + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + upstreamReconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + fetchCalls += 1; + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + }, + }); + + const firstSocket = new FakeSocket(); + const secondSocket = new FakeSocket(); + runtime.wsServer.emit('connection', firstSocket, { url: '/api/global/event/ws' }); + runtime.wsServer.emit('connection', secondSocket, { url: '/api/global/event/ws' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(fetchCalls).toBe(1); + expect(firstSocket.sent).toContainEqual({ type: 'ready', scope: 'global' }); + expect(secondSocket.sent).toContainEqual({ type: 'ready', scope: 'global' }); + expect(firstSocket.sent).toContainEqual({ + type: 'event', + payload: { type: 'server.connected', properties: {} }, + eventId: 'evt-1', + directory: 'global', + }); + expect(secondSocket.sent).toContainEqual({ + type: 'event', + payload: { type: 'server.connected', properties: {} }, + eventId: 'evt-1', + directory: 'global', + }); + + firstSocket.close(); + secondSocket.close(); + await runtime.close(); + }); + + it('replays buffered global events after a reconnecting client Last-Event-ID', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let fetchCalls = 0; + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + upstreamReconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + fetchCalls += 1; + if (fetchCalls === 1) { + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + 'id: evt-2\ndata: {"type":"session.updated","properties":{"directory":"/tmp/project"}}\n\n', + ], + }); + } + + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [], + }); + }, + }); + + const firstSocket = new FakeSocket(); + runtime.wsServer.emit('connection', firstSocket, { url: '/api/global/event/ws' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + firstSocket.close(); + + const secondSocket = new FakeSocket(); + runtime.wsServer.emit('connection', secondSocket, { url: '/api/global/event/ws?lastEventId=evt-1' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(secondSocket.sent).toContainEqual({ type: 'ready', scope: 'global' }); + expect(secondSocket.sent).toContainEqual({ + type: 'event', + payload: { type: 'session.updated', properties: { directory: '/tmp/project' } }, + eventId: 'evt-2', + directory: '/tmp/project', + }); + + secondSocket.close(); + await runtime.close(); + }); + + it('keeps directory websocket streams on separate upstream readers', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + const fetchUrls = []; + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + upstreamReconnectDelayMs: 0, + fetchImpl: async (url, options) => { + fetchUrls.push(url); + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + }, + }); + + const firstSocket = new FakeSocket(); + const secondSocket = new FakeSocket(); + runtime.wsServer.emit('connection', firstSocket, { url: '/api/event/ws?directory=/tmp/one' }); + runtime.wsServer.emit('connection', secondSocket, { url: '/api/event/ws?directory=/tmp/two' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(fetchUrls).toHaveLength(2); + expect(new URL(fetchUrls[0]).searchParams.get('directory')).toBe('/tmp/one'); + expect(new URL(fetchUrls[1]).searchParams.get('directory')).toBe('/tmp/two'); + expect(firstSocket.sent).toContainEqual({ type: 'ready', scope: 'directory' }); + expect(secondSocket.sent).toContainEqual({ type: 'ready', scope: 'directory' }); + + firstSocket.close(); + secondSocket.close(); + await runtime.close(); + }); + + it('closes the websocket and triggers health check on initial upstream unavailable response', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let triggerHealthCheckCalls = 0; + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + triggerHealthCheck: () => { + triggerHealthCheckCalls += 1; + }, + upstreamReconnectDelayMs: 0, + fetchImpl: async () => ({ + ok: false, + status: 503, + body: null, + }), + }); + + const socket = new FakeSocket(); + runtime.wsServer.emit('connection', socket, { url: '/api/global/event/ws' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(socket.sent).toEqual([ + { + type: 'error', + message: 'OpenCode event stream unavailable (503)', + }, + ]); + expect(socket.closeCalls).toEqual([ + { + code: 1011, + reason: 'OpenCode event stream unavailable', + }, + ]); + expect(triggerHealthCheckCalls).toBe(1); + expect(wsClients.size).toBe(0); + + await runtime.close(); + }); + + it('closes the websocket without health check when OpenCode URL cannot be built', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let triggerHealthCheckCalls = 0; + let fetchCalls = 0; + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl() { + throw new Error('missing OpenCode port'); + }, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + triggerHealthCheck: () => { + triggerHealthCheckCalls += 1; + }, + upstreamReconnectDelayMs: 0, + fetchImpl: async () => { + fetchCalls += 1; + throw new Error('fetch should not be called'); + }, + }); + + const socket = new FakeSocket(); + runtime.wsServer.emit('connection', socket, { url: '/api/global/event/ws' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(socket.sent).toEqual([ + { + type: 'error', + message: 'OpenCode service unavailable', + }, + ]); + expect(socket.closeCalls).toEqual([ + { + code: 1011, + reason: 'OpenCode service unavailable', + }, + ]); + expect(fetchCalls).toBe(0); + expect(triggerHealthCheckCalls).toBe(0); + + await runtime.close(); + }); + it('reconnects a stalled upstream SSE stream and resumes from the last event id', async () => { const server = new EventEmitter(); const wsClients = new Set(); @@ -192,4 +459,54 @@ describe('message stream websocket runtime', () => { socket.close(); await runtime.close(); }); + + it('keeps synthetic event processing on forwarded upstream events', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload(payload, emitSynthetic) { + if (payload.type === 'session.updated') { + emitSynthetic({ type: 'openchamber:session-status', sessionID: 'ses_1' }); + } + }, + wsClients, + upstreamReconnectDelayMs: 0, + fetchImpl: async (_url, options) => createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"session.updated","properties":{"directory":"/tmp/project"}}\n\n', + ], + }), + }); + + const socket = new FakeSocket(); + runtime.wsServer.emit('connection', socket, { url: '/api/global/event/ws' }); + + await new Promise((resolve) => setTimeout(resolve, 5)); + + expect(socket.sent).toContainEqual({ + type: 'event', + payload: { type: 'session.updated', properties: { directory: '/tmp/project' } }, + eventId: 'evt-1', + directory: '/tmp/project', + }); + expect(socket.sent).toContainEqual({ + type: 'event', + payload: { type: 'openchamber:session-status', sessionID: 'ses_1' }, + directory: 'global', + }); + + socket.close(); + await runtime.close(); + }); }); diff --git a/packages/web/server/lib/event-stream/upstream-reader.js b/packages/web/server/lib/event-stream/upstream-reader.js new file mode 100644 index 00000000..253a1e9c --- /dev/null +++ b/packages/web/server/lib/event-stream/upstream-reader.js @@ -0,0 +1,204 @@ +import { parseSseEventEnvelope } from './protocol.js'; + +export const DEFAULT_UPSTREAM_STALL_TIMEOUT_MS = 20_000; +export const DEFAULT_UPSTREAM_RECONNECT_DELAY_MS = 250; + +function waitForReconnectDelay(ms, signal) { + if (signal?.aborted) { + return Promise.resolve(); + } + + return new Promise((resolve) => { + const timeout = setTimeout(resolve, Math.max(0, ms)); + signal?.addEventListener('abort', () => { + clearTimeout(timeout); + resolve(); + }, { once: true }); + }); +} + +function normalizeHeaders(headers) { + if (!headers || typeof headers !== 'object') { + return {}; + } + + return { ...headers }; +} + +export function createUpstreamSseReader({ + buildUrl, + getHeaders = () => ({}), + fetchImpl = fetch, + parseBlock = parseSseEventEnvelope, + initialLastEventId = '', + signal, + stallTimeoutMs = DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + reconnectDelayMs = DEFAULT_UPSTREAM_RECONNECT_DELAY_MS, + onEvent, + onConnect, + onDisconnect, + onError, +}) { + let running = null; + let stopped = false; + let activeController = null; + let lastEventId = typeof initialLastEventId === 'string' ? initialLastEventId : ''; + + const stop = () => { + stopped = true; + if (activeController && !activeController.signal.aborted) { + activeController.abort(); + } + }; + + signal?.addEventListener('abort', stop, { once: true }); + + const start = () => { + if (running) { + return running; + } + + stopped = false; + running = (async () => { + while (!stopped && !signal?.aborted) { + const controller = new AbortController(); + activeController = controller; + const abortActive = () => controller.abort(); + signal?.addEventListener('abort', abortActive, { once: true }); + + let abortReason = null; + let stallTimer = null; + const clearStallTimer = () => { + if (stallTimer) { + clearTimeout(stallTimer); + stallTimer = null; + } + }; + const resetStallTimer = () => { + clearStallTimer(); + if (stallTimeoutMs <= 0) { + return; + } + + stallTimer = setTimeout(() => { + abortReason = 'upstream_stalled'; + controller.abort(); + }, stallTimeoutMs); + }; + + try { + const url = buildUrl(); + const headers = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + ...normalizeHeaders(getHeaders()), + }; + if (lastEventId) { + headers['Last-Event-ID'] = lastEventId; + } + + const response = await fetchImpl(url.toString(), { + headers, + signal: controller.signal, + }); + + if (!response?.ok || !response.body) { + onError?.({ + type: 'upstream_unavailable', + status: response?.status ?? 0, + response, + }); + await waitForReconnectDelay(reconnectDelayMs, signal); + continue; + } + + onConnect?.({ response, lastEventId }); + + const decoder = new TextDecoder(); + const reader = response.body.getReader(); + let buffer = ''; + + resetStallTimer(); + + while (!stopped && !signal?.aborted) { + const { value, done } = await reader.read(); + if (done) { + break; + } + + resetStallTimer(); + buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n'); + + let separatorIndex = buffer.indexOf('\n\n'); + while (separatorIndex !== -1 && !stopped && !signal?.aborted) { + const block = buffer.slice(0, separatorIndex); + buffer = buffer.slice(separatorIndex + 2); + const envelope = parseBlock(block); + if (envelope?.payload) { + if (typeof envelope.eventId === 'string' && envelope.eventId.length > 0) { + lastEventId = envelope.eventId; + } + onEvent?.({ + block, + envelope, + payload: envelope.payload, + eventId: envelope.eventId, + directory: envelope.directory, + }); + } + separatorIndex = buffer.indexOf('\n\n'); + } + } + + if (!stopped && !signal?.aborted && buffer.trim().length > 0) { + const block = buffer.trim(); + const envelope = parseBlock(block); + if (envelope?.payload) { + if (typeof envelope.eventId === 'string' && envelope.eventId.length > 0) { + lastEventId = envelope.eventId; + } + onEvent?.({ + block, + envelope, + payload: envelope.payload, + eventId: envelope.eventId, + directory: envelope.directory, + }); + } + } + } catch (error) { + if (!stopped && !signal?.aborted && abortReason !== 'upstream_stalled') { + onError?.({ + type: 'stream_error', + error, + }); + } + } finally { + clearStallTimer(); + signal?.removeEventListener('abort', abortActive); + if (activeController === controller) { + activeController = null; + } + onDisconnect?.({ reason: abortReason ?? (stopped || signal?.aborted ? 'stopped' : 'closed') }); + } + + if (!stopped && !signal?.aborted) { + await waitForReconnectDelay(reconnectDelayMs, signal); + } + } + })().finally(() => { + running = null; + }); + + return running; + }; + + return { + start, + stop, + getLastEventId() { + return lastEventId; + }, + }; +} diff --git a/packages/web/server/lib/event-stream/upstream-reader.test.js b/packages/web/server/lib/event-stream/upstream-reader.test.js new file mode 100644 index 00000000..652dd340 --- /dev/null +++ b/packages/web/server/lib/event-stream/upstream-reader.test.js @@ -0,0 +1,159 @@ +import { describe, expect, it } from 'vitest'; + +import { createUpstreamSseReader } from './upstream-reader.js'; + +function createSseResponse({ blocks = [], signal, holdOpen = false }) { + const encoder = new TextEncoder(); + let index = 0; + + return { + ok: true, + status: 200, + body: { + getReader() { + return { + async read() { + if (index < blocks.length) { + return { value: encoder.encode(blocks[index++]), done: false }; + } + + if (!holdOpen) { + return { value: undefined, done: true }; + } + + return new Promise((_resolve, reject) => { + const onAbort = () => { + signal.removeEventListener('abort', onAbort); + const error = new Error('Aborted'); + error.name = 'AbortError'; + reject(error); + }; + signal.addEventListener('abort', onAbort, { once: true }); + }); + }, + }; + }, + }, + }; +} + +describe('createUpstreamSseReader', () => { + it('emits parsed events and tracks the latest event id', async () => { + const events = []; + let reader; + + reader = createUpstreamSseReader({ + buildUrl: () => 'http://127.0.0.1:4096/global/event', + reconnectDelayMs: 0, + fetchImpl: async (_url, options) => createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-1\r\ndata: {"type":"server.connected","properties":{"directory":"/tmp/project"}}\r\n\r\n', + ], + }), + onEvent(event) { + events.push(event); + reader.stop(); + }, + }); + + await reader.start(); + + expect(events).toHaveLength(1); + expect(events[0].eventId).toBe('evt-1'); + expect(events[0].directory).toBe('/tmp/project'); + expect(events[0].payload).toEqual({ + type: 'server.connected', + properties: { + directory: '/tmp/project', + }, + }); + expect(reader.getLastEventId()).toBe('evt-1'); + }); + + it('reconnects a stalled stream with Last-Event-ID', async () => { + const fetchLastEventIds = []; + const events = []; + let attempt = 0; + let reader; + + reader = createUpstreamSseReader({ + buildUrl: () => 'http://127.0.0.1:4096/global/event', + stallTimeoutMs: 10, + reconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + fetchLastEventIds.push(options.headers['Last-Event-ID'] ?? null); + attempt += 1; + + if (attempt === 1) { + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + } + + return createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-2\ndata: {"type":"session.updated","properties":{}}\n\n', + ], + }); + }, + onEvent(event) { + events.push(event.eventId); + if (event.eventId === 'evt-2') { + reader.stop(); + } + }, + }); + + await reader.start(); + + expect(events).toEqual(['evt-1', 'evt-2']); + expect(fetchLastEventIds.slice(0, 2)).toEqual([null, 'evt-1']); + expect(reader.getLastEventId()).toBe('evt-2'); + }); + + it('reports unavailable upstream responses and continues reconnecting until stopped', async () => { + const errors = []; + let attempt = 0; + let reader; + + reader = createUpstreamSseReader({ + buildUrl: () => 'http://127.0.0.1:4096/global/event', + reconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + attempt += 1; + if (attempt === 1) { + return { ok: false, status: 503, body: null }; + } + + return createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + }, + onError(error) { + errors.push(error); + }, + onEvent() { + reader.stop(); + }, + }); + + await reader.start(); + + expect(errors).toEqual([ + expect.objectContaining({ + type: 'upstream_unavailable', + status: 503, + }), + ]); + expect(attempt).toBe(2); + }); +}); diff --git a/packages/web/server/lib/opencode/DOCUMENTATION.md b/packages/web/server/lib/opencode/DOCUMENTATION.md index f58fdce2..fd74239a 100644 --- a/packages/web/server/lib/opencode/DOCUMENTATION.md +++ b/packages/web/server/lib/opencode/DOCUMENTATION.md @@ -335,10 +335,16 @@ This module provides OpenCode server integration utilities for the web server ru - OpenCode readiness gate for proxied `/api` requests ## Public exports (watcher.js) -- `createOpenCodeWatcherRuntime(dependencies)`: creates global event watcher runtime. +- `createOpenCodeWatcherRuntime(dependencies)`: creates global event watcher runtime backed by the shared upstream SSE reader. - Returned API: - `start()` - `stop()` +- Behavior: + - Waits for OpenCode readiness before attaching the watcher. + - In production wiring, subscribes to the shared global message-stream hub instead of opening its own `/global/event` connection. + - Can still create its own `/global/event` reader when no shared hub is provided, which keeps module tests and isolated reuse simple. + - Reuses event-stream parsing, `Last-Event-ID`, stall timeout, and reconnect behavior. + - Forwards unwrapped global event payloads into notification/session side effects. ## Storage and configuration - Provider auth: `~/.local/share/opencode/auth.json`. diff --git a/packages/web/server/lib/opencode/startup-pipeline-runtime.js b/packages/web/server/lib/opencode/startup-pipeline-runtime.js index 5860a3e3..c17c5be9 100644 --- a/packages/web/server/lib/opencode/startup-pipeline-runtime.js +++ b/packages/web/server/lib/opencode/startup-pipeline-runtime.js @@ -20,6 +20,7 @@ export const createStartupPipelineRuntime = (dependencies) => { rejectWebSocketUpgrade, buildOpenCodeUrl, getOpenCodeAuthHeaders, + globalEventHub, processForwardedEventPayload, messageStreamWsClients, triggerHealthCheck, @@ -75,6 +76,7 @@ export const createStartupPipelineRuntime = (dependencies) => { rejectWebSocketUpgrade, buildOpenCodeUrl, getOpenCodeAuthHeaders, + globalEventHub, processForwardedEventPayload, wsClients: messageStreamWsClients, triggerHealthCheck, diff --git a/packages/web/server/lib/opencode/watcher.js b/packages/web/server/lib/opencode/watcher.js index c8d20091..c973c026 100644 --- a/packages/web/server/lib/opencode/watcher.js +++ b/packages/web/server/lib/opencode/watcher.js @@ -1,4 +1,4 @@ -import { createOpencodeClient } from '@opencode-ai/sdk/v2'; +import { createUpstreamSseReader } from '../event-stream/upstream-reader.js'; export const createOpenCodeWatcherRuntime = (deps) => { const { @@ -6,9 +6,16 @@ export const createOpenCodeWatcherRuntime = (deps) => { buildOpenCodeUrl, getOpenCodeAuthHeaders, onPayload, + fetchImpl = fetch, + upstreamStallTimeoutMs, + upstreamReconnectDelayMs = 1000, + globalEventHub = null, } = deps; let abortController = null; + let reader = null; + let unsubscribeEvent = null; + let unsubscribeStatus = null; const unwrapGlobalEventPayload = (eventData) => { if (!eventData || typeof eventData !== 'object') { @@ -32,50 +39,56 @@ export const createOpenCodeWatcherRuntime = (deps) => { abortController = new AbortController(); const signal = abortController.signal; - let attempt = 0; - const run = async () => { - while (!signal.aborted) { - attempt += 1; - try { - const baseUrl = buildOpenCodeUrl('/', '').replace(/\/$/, ''); - const client = createOpencodeClient({ - baseUrl, - headers: getOpenCodeAuthHeaders(), - }); - - const result = await client.global.event({ - signal, - sseMaxRetryAttempts: 0, - onSseEvent: (event) => { - const payload = unwrapGlobalEventPayload(event.data); - if (!payload || typeof payload !== 'object') { - return; - } - onPayload(payload); - }, - }); - - console.log('[PushWatcher] connected'); - - for await (const _ of result.stream) { - void _; - if (signal.aborted) { - break; - } - } - } catch (error) { - if (signal.aborted) { - return; - } - console.warn('[PushWatcher] disconnected', error?.message ?? error); + if (globalEventHub) { + unsubscribeEvent = globalEventHub.subscribeEvent((event) => { + const payload = unwrapGlobalEventPayload(event.payload); + if (!payload || typeof payload !== 'object') { + return; } + onPayload(payload); + }); + unsubscribeStatus = globalEventHub.subscribeStatus((status) => { + if (signal.aborted) { + return; + } + if (status.type === 'connect') { + console.log('[PushWatcher] connected'); + return; + } + if (status.type === 'error' || status.type === 'initial-error') { + console.warn('[PushWatcher] disconnected', status.error?.error?.message ?? status.error?.message ?? status.error); + } + }); + globalEventHub.start(); + return; + } - const backoffMs = Math.min(1000 * Math.pow(2, Math.min(attempt, 5)), 30000); - await new Promise((resolve) => setTimeout(resolve, backoffMs)); - } - }; + reader = createUpstreamSseReader({ + signal, + buildUrl: () => buildOpenCodeUrl('/global/event', ''), + getHeaders: getOpenCodeAuthHeaders, + fetchImpl, + stallTimeoutMs: upstreamStallTimeoutMs, + reconnectDelayMs: upstreamReconnectDelayMs, + onConnect() { + console.log('[PushWatcher] connected'); + }, + onEvent(event) { + const payload = unwrapGlobalEventPayload(event.payload); + if (!payload || typeof payload !== 'object') { + return; + } + onPayload(payload); + }, + onError(error) { + if (signal.aborted) { + return; + } + console.warn('[PushWatcher] disconnected', error?.error?.message ?? error?.message ?? error); + }, + }); - void run(); + void reader.start(); }; const stop = () => { @@ -84,8 +97,14 @@ export const createOpenCodeWatcherRuntime = (deps) => { } try { abortController.abort(); + reader?.stop(); + unsubscribeEvent?.(); + unsubscribeStatus?.(); } catch { } + reader = null; + unsubscribeEvent = null; + unsubscribeStatus = null; abortController = null; }; diff --git a/packages/web/server/lib/opencode/watcher.test.js b/packages/web/server/lib/opencode/watcher.test.js new file mode 100644 index 00000000..f7c65ad7 --- /dev/null +++ b/packages/web/server/lib/opencode/watcher.test.js @@ -0,0 +1,239 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { createGlobalMessageStreamHub } from '../event-stream/global-hub.js'; +import { createOpenCodeWatcherRuntime } from './watcher.js'; + +function createSseResponse({ blocks = [], signal, holdOpen = false }) { + const encoder = new TextEncoder(); + let index = 0; + + return { + ok: true, + status: 200, + body: { + getReader() { + return { + async read() { + if (index < blocks.length) { + return { value: encoder.encode(blocks[index++]), done: false }; + } + + if (!holdOpen) { + return { value: undefined, done: true }; + } + + return new Promise((_resolve, reject) => { + const onAbort = () => { + signal.removeEventListener('abort', onAbort); + const error = new Error('Aborted'); + error.name = 'AbortError'; + reject(error); + }; + signal.addEventListener('abort', onAbort, { once: true }); + }); + }, + }; + }, + }, + }; +} + +describe('createOpenCodeWatcherRuntime', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('waits for OpenCode readiness and forwards unwrapped global SSE payloads', async () => { + vi.spyOn(console, 'log').mockImplementation(() => {}); + const payloads = []; + const fetchCalls = []; + + const watcher = createOpenCodeWatcherRuntime({ + waitForOpenCodePort: async () => { + await new Promise((resolve) => setTimeout(resolve, 1)); + }, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({ Authorization: 'Bearer test-token' }), + onPayload(payload) { + payloads.push(payload); + watcher.stop(); + }, + fetchImpl: async (url, options) => { + fetchCalls.push({ url, headers: options.headers }); + return createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-1\ndata: {"directory":"/tmp/project","payload":{"type":"session.updated","properties":{"sessionID":"ses_1"}}}\n\n', + ], + }); + }, + }); + + await watcher.start(); + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(fetchCalls).toEqual([ + { + url: 'http://127.0.0.1:4096/global/event', + headers: { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + Authorization: 'Bearer test-token', + }, + }, + ]); + expect(payloads).toEqual([ + { + type: 'session.updated', + properties: { + sessionID: 'ses_1', + }, + }, + ]); + }); + + it('resumes watcher reconnects with Last-Event-ID after a stalled upstream stream', async () => { + vi.spyOn(console, 'log').mockImplementation(() => {}); + const fetchLastEventIds = []; + const payloads = []; + let attempt = 0; + + const watcher = createOpenCodeWatcherRuntime({ + waitForOpenCodePort: async () => {}, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + onPayload(payload) { + payloads.push(payload.type); + if (payload.type === 'session.updated') { + watcher.stop(); + } + }, + fetchImpl: async (_url, options) => { + fetchLastEventIds.push(options.headers['Last-Event-ID'] ?? null); + attempt += 1; + + if (attempt === 1) { + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + } + + return createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-2\ndata: {"type":"session.updated","properties":{}}\n\n', + ], + }); + }, + upstreamStallTimeoutMs: 10, + upstreamReconnectDelayMs: 0, + }); + + await watcher.start(); + await new Promise((resolve) => setTimeout(resolve, 30)); + + expect(payloads).toEqual(['server.connected', 'session.updated']); + expect(fetchLastEventIds.slice(0, 2)).toEqual([null, 'evt-1']); + }); + + it('subscribes to a shared global event hub instead of opening its own upstream stream', async () => { + vi.spyOn(console, 'log').mockImplementation(() => {}); + const payloads = []; + let hubFetchCalls = 0; + let watcherFetchCalls = 0; + + const globalEventHub = createGlobalMessageStreamHub({ + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + upstreamReconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + hubFetchCalls += 1; + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"payload":{"type":"session.updated","properties":{"sessionID":"ses_1"}}}\n\n', + ], + }); + }, + }); + + const watcher = createOpenCodeWatcherRuntime({ + waitForOpenCodePort: async () => {}, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + globalEventHub, + onPayload(payload) { + payloads.push(payload); + watcher.stop(); + }, + fetchImpl: async () => { + watcherFetchCalls += 1; + throw new Error('watcher fetch should not be called'); + }, + }); + + await watcher.start(); + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(hubFetchCalls).toBe(1); + expect(watcherFetchCalls).toBe(0); + expect(payloads).toEqual([ + { + type: 'session.updated', + properties: { + sessionID: 'ses_1', + }, + }, + ]); + }); + + it('does not stop a shared global event hub when the watcher stops', async () => { + const events = new Set(); + const statuses = new Set(); + let startCalls = 0; + let stopCalls = 0; + + const globalEventHub = { + start() { + startCalls += 1; + }, + stop() { + stopCalls += 1; + }, + subscribeEvent(subscriber) { + events.add(subscriber); + return () => { + events.delete(subscriber); + }; + }, + subscribeStatus(subscriber) { + statuses.add(subscriber); + return () => { + statuses.delete(subscriber); + }; + }, + }; + + const watcher = createOpenCodeWatcherRuntime({ + waitForOpenCodePort: async () => {}, + buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`, + getOpenCodeAuthHeaders: () => ({}), + globalEventHub, + onPayload() {}, + }); + + await watcher.start(); + watcher.stop(); + + expect(startCalls).toBe(1); + expect(stopCalls).toBe(0); + expect(events.size).toBe(0); + expect(statuses.size).toBe(0); + }); +});