From 6bcea3eb4da04ee6cd913a37925a673290eb1b89 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Wed, 24 Dec 2025 03:25:52 +0200 Subject: [PATCH] feat(web): sse event streaming and activity tracking --- packages/ui/src/hooks/useEventStream.ts | 22 +-- packages/web/server/index.js | 178 ++++++++++++++++++++++++ 2 files changed, 191 insertions(+), 9 deletions(-) diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 09077cd7..9901cb1e 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -407,7 +407,6 @@ export const useEventStream = () => { break; } case 'openchamber:session-activity': { - if (!isDesktopRuntimeRef.current) break; const sessionId = typeof props.sessionId === 'string' ? props.sessionId : null; const phase = typeof props.phase === 'string' ? props.phase : null; if (sessionId && (phase === 'idle' || phase === 'busy' || phase === 'cooldown')) { @@ -804,16 +803,21 @@ export const useEventStream = () => { completeStreamingMessage(sessionId, messageId); + // For web/vscode: trigger cooldown only when assistant message has finish === "stop" + // This matches the desktop backend logic in session_activity.rs if (!isDesktopRuntimeRef.current) { - const rawCompletedSessionId = (message as { sessionID?: string }).sessionID; - const completedSessionId: string = - typeof rawCompletedSessionId === 'string' && rawCompletedSessionId.length > 0 - ? rawCompletedSessionId - : sessionId; + const finish = (message as { finish?: string }).finish; + if (finish === 'stop') { + const rawCompletedSessionId = (message as { sessionID?: string }).sessionID; + const completedSessionId: string = + typeof rawCompletedSessionId === 'string' && rawCompletedSessionId.length > 0 + ? rawCompletedSessionId + : sessionId; - const currentPhase = sessionActivityPhaseRef.current.get(completedSessionId); - if (currentPhase === 'busy') { - updateSessionActivityPhase(completedSessionId, 'cooldown'); + const currentPhase = sessionActivityPhaseRef.current.get(completedSessionId); + if (currentPhase === 'busy') { + updateSessionActivityPhase(completedSessionId, 'cooldown'); + } } } diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 85c5dac8..918f1d8a 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -677,6 +677,61 @@ function buildOpenCodeUrl(path, prefixOverride) { return `http://localhost:${openCodePort}${fullPath}`; } +function parseSseDataPayload(block) { + if (!block || typeof block !== 'string') { + return null; + } + const dataLines = block + .split('\n') + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).replace(/^\s/, '')); + + if (dataLines.length === 0) { + return null; + } + + const payloadText = dataLines.join('\n').trim(); + if (!payloadText) { + return null; + } + + try { + return JSON.parse(payloadText); + } catch { + return null; + } +} + +function deriveSessionActivity(payload) { + if (!payload || typeof payload !== 'object') { + return null; + } + + if (payload.type === 'session.status') { + const status = payload.properties?.status; + const sessionId = payload.properties?.sessionID; + const statusType = status?.type; + + if (typeof sessionId === 'string' && sessionId.length > 0 && typeof statusType === 'string') { + const phase = statusType === 'busy' || statusType === 'retry' ? 'busy' : 'idle'; + return { sessionId, phase }; + } + } + + if (payload.type === 'session.idle') { + const sessionId = payload.properties?.sessionID; + if (typeof sessionId === 'string' && sessionId.length > 0) { + return { sessionId, phase: 'idle' }; + } + } + + return null; +} + +function writeSseEvent(res, payload) { + res.write(`data: ${JSON.stringify(payload)}\n\n`); +} + function extractApiPrefixFromUrl(urlString, expectedSuffix) { if (!urlString) { return null; @@ -1764,6 +1819,129 @@ async function main(options = {}) { } }); + app.get('/api/event', async (req, res) => { + if (!openCodePort) { + return res.status(503).json({ error: 'OpenCode service unavailable' }); + } + + if (!openCodeApiPrefixDetected) { + try { + await detectOpenCodeApiPrefix(); + } catch { + // ignore detection failures + } + } + + let targetUrl; + try { + const prefix = openCodeApiPrefixDetected ? openCodeApiPrefix : ''; + targetUrl = new URL(buildOpenCodeUrl('/event', prefix)); + } catch (error) { + return res.status(503).json({ error: 'OpenCode service unavailable' }); + } + + const directoryParam = Array.isArray(req.query.directory) + ? req.query.directory[0] + : req.query.directory; + if (typeof directoryParam === 'string' && directoryParam.trim().length > 0) { + targetUrl.searchParams.set('directory', directoryParam.trim()); + } + + const headers = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive' + }; + + const lastEventId = req.header('Last-Event-ID'); + if (typeof lastEventId === 'string' && lastEventId.length > 0) { + headers['Last-Event-ID'] = lastEventId; + } + + const controller = new AbortController(); + const cleanup = () => { + if (!controller.signal.aborted) { + controller.abort(); + } + }; + + req.on('close', cleanup); + req.on('error', cleanup); + + let upstream; + try { + upstream = await fetch(targetUrl.toString(), { + headers, + signal: controller.signal, + }); + } catch (error) { + return res.status(502).json({ error: 'Failed to connect to OpenCode event stream' }); + } + + if (!upstream.ok || !upstream.body) { + return res.status(502).json({ error: `OpenCode event stream unavailable (${upstream.status})` }); + } + + res.setHeader('Content-Type', 'text/event-stream'); + res.setHeader('Cache-Control', 'no-cache'); + res.setHeader('Connection', 'keep-alive'); + res.setHeader('X-Accel-Buffering', 'no'); + + if (typeof res.flushHeaders === 'function') { + res.flushHeaders(); + } + + const decoder = new TextDecoder(); + const reader = upstream.body.getReader(); + let buffer = ''; + + const forwardBlock = (block) => { + if (!block) return; + res.write(`${block}\n\n`); + const payload = parseSseDataPayload(block); + const activity = deriveSessionActivity(payload); + if (activity) { + writeSseEvent(res, { + type: 'openchamber:session-activity', + properties: { + sessionId: activity.sessionId, + phase: activity.phase, + } + }); + } + }; + + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n'); + + let separatorIndex; + while ((separatorIndex = buffer.indexOf('\n\n')) !== -1) { + const block = buffer.slice(0, separatorIndex); + buffer = buffer.slice(separatorIndex + 2); + forwardBlock(block); + } + } + + if (buffer.trim().length > 0) { + forwardBlock(buffer.trim()); + } + } catch (error) { + if (!controller.signal.aborted) { + console.warn('SSE proxy stream error:', error); + } + } finally { + cleanup(); + try { + res.end(); + } catch { + // ignore + } + } + }); + app.get('/api/config/settings', async (_req, res) => { try { const settings = await readSettingsFromDisk();