From 45b3d1d9201c6598c5aa813521c0552f4d7a48cd Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Sun, 4 Jan 2026 10:50:35 +0200 Subject: [PATCH] feat(event-stream): add global event stream and activity tracking Add global event stream endpoint for cross-session activity monitoring Support multiple concurrent SSE subscriptions with scoped connections Enhance session activity detection from various event types --- packages/ui/src/hooks/useEventStream.ts | 310 ++++++++++++++++-- packages/ui/src/lib/opencode/client.ts | 32 +- .../vscode/src/AgentManagerPanelProvider.ts | 129 +++++++- packages/vscode/src/ChatViewProvider.ts | 129 +++++++- packages/web/server/index.js | 139 +++++++- 5 files changed, 686 insertions(+), 53 deletions(-) diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 545961af..dbfb959e 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -127,6 +127,31 @@ export const useEventStream = () => { return undefined; }, [activeSessionDirectory, fallbackDirectory]); + const normalizeDirectory = React.useCallback((value: string | null | undefined): string | null => { + if (typeof value !== 'string') return null; + const trimmed = value.trim(); + if (!trimmed) return null; + const normalized = trimmed.replace(/\\/g, '/'); + return normalized.length > 1 ? normalized.replace(/\/+$/, '') : normalized; + }, []); + + const resolveSessionDirectoryForStatus = React.useCallback( + (sessionId: string | null | undefined): string | null => { + if (!sessionId) return null; + try { + const metadata = getWorktreeMetadata?.(sessionId); + const metaPath = normalizeDirectory(metadata?.path ?? null); + if (metaPath) return metaPath; + } catch { + // ignored + } + + const record = sessions.find((entry) => entry.id === sessionId); + return normalizeDirectory((record as { directory?: string | null })?.directory ?? null); + }, + [getWorktreeMetadata, normalizeDirectory, sessions] + ); + const setEventStreamStatus = useUIStore((state) => state.setEventStreamStatus); const lastStatusRef = React.useRef<{ status: EventStreamStatus; hint: string | null } | null>(null); @@ -240,6 +265,7 @@ export const useEventStream = () => { const staleCheckIntervalRef = React.useRef(null); const lastEventTimestampRef = React.useRef(Date.now()); const isDesktopRuntimeRef = React.useRef(false); + const activityStreamAbortControllerRef = React.useRef(null); const maybeBootstrapIfStale = React.useCallback( (reason: string) => { @@ -266,6 +292,8 @@ export const useEventStream = () => { const sessionStatusLastRefreshAtRef = React.useRef(0); const sessionStatusRefreshInFlightRef = React.useRef | null>(null); const currentSessionIdRef = React.useRef(currentSessionId); + const previousSessionIdRef = React.useRef(null); + const previousSessionDirectoryRef = React.useRef(null); React.useEffect(() => { currentSessionIdRef.current = currentSessionId; }, [currentSessionId]); @@ -361,28 +389,6 @@ export const useEventStream = () => { } sessionStatusLastRefreshAtRef.current = now; - const normalizeDirectory = (value: string | null | undefined): string | null => { - if (typeof value !== 'string') return null; - const trimmed = value.trim(); - if (!trimmed) return null; - const normalized = trimmed.replace(/\\/g, '/'); - return normalized.length > 1 ? normalized.replace(/\/+$/, '') : normalized; - }; - - const resolveSessionDirectoryForStatus = (sessionId: string): string | null => { - try { - const metadata = getWorktreeMetadata?.(sessionId); - const metaPath = normalizeDirectory(metadata?.path ?? null); - if (metaPath) return metaPath; - } catch { - // ignored - } - - const record = sessions.find((entry) => entry.id === sessionId); - const recordPath = normalizeDirectory((record as { directory?: string | null })?.directory ?? null); - return recordPath; - }; - const applyStatusMap = (statusMap: Record) => { Object.entries(statusMap).forEach(([sessionId, raw]) => { if (!sessionId || !raw) return; @@ -439,7 +445,109 @@ export const useEventStream = () => { sessionStatusRefreshInFlightRef.current = task; return task; - }, [effectiveDirectory, getWorktreeMetadata, sessions, updateSessionActivityPhase]); + }, [effectiveDirectory, normalizeDirectory, resolveSessionDirectoryForStatus, sessions, updateSessionActivityPhase]); + + React.useEffect(() => { + const nextSessionId = currentSessionId ?? null; + const prevSessionId = previousSessionIdRef.current; + const nextDirectory = resolveSessionDirectoryForStatus(nextSessionId); + const prevDirectory = previousSessionDirectoryRef.current; + + if (prevSessionId && nextSessionId && prevSessionId !== nextSessionId) { + if (prevDirectory && nextDirectory && prevDirectory !== nextDirectory) { + void refreshSessionActivityStatus(); + } + } + + previousSessionIdRef.current = nextSessionId; + previousSessionDirectoryRef.current = nextDirectory; + }, [currentSessionId, refreshSessionActivityStatus, resolveSessionDirectoryForStatus]); + + const handleActivityEvent = React.useCallback((event: EventData) => { + if (!event?.type) return; + + const props = (event.properties ?? {}) as Record; + + if (event.type === 'openchamber:session-activity') { + const sessionId = + typeof props.sessionId === 'string' + ? props.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')) { + updateSessionActivityPhase(sessionId, phase); + requestSessionListRefresh(); + } + return; + } + + if (event.type === 'session.status') { + const sessionId = + typeof props.sessionID === 'string' + ? props.sessionID + : typeof props.sessionId === 'string' + ? props.sessionId + : null; + const statusObj = + typeof props.status === 'object' && props.status !== null + ? (props.status as Record) + : null; + const statusType = typeof statusObj?.type === 'string' ? (statusObj.type as string) : null; + + if (sessionId && statusType) { + updateSessionActivityPhase( + sessionId, + statusType === 'busy' || statusType === 'retry' ? 'busy' : 'idle', + ); + requestSessionListRefresh(); + } + return; + } + + if (event.type === 'session.idle') { + const sessionId = + typeof props.sessionID === 'string' + ? props.sessionID + : typeof props.sessionId === 'string' + ? props.sessionId + : null; + if (sessionId) { + updateSessionActivityPhase(sessionId, 'idle'); + requestSessionListRefresh(); + } + return; + } + + if (event.type === 'message.updated' || event.type === 'message.part.updated') { + const messageInfo = + typeof props.info === 'object' && props.info !== null ? (props.info as Record) : props; + + const sessionId = + typeof (messageInfo as { sessionID?: unknown }).sessionID === 'string' + ? (messageInfo as { sessionID?: string }).sessionID + : typeof (messageInfo as { sessionId?: unknown }).sessionId === 'string' + ? (messageInfo as { sessionId?: string }).sessionId + : typeof props.sessionID === 'string' + ? (props.sessionID as string) + : typeof props.sessionId === 'string' + ? (props.sessionId as string) + : null; + + const role = (messageInfo as { role?: unknown }).role; + const finish = (messageInfo as { finish?: unknown }).finish; + + if (sessionId && role === 'assistant' && finish === 'stop') { + const currentPhase = useSessionStore.getState().sessionActivityPhase?.get(sessionId); + if (currentPhase === 'busy') { + updateSessionActivityPhase(sessionId, 'cooldown'); + requestSessionListRefresh(); + } + } + return; + } + }, [requestSessionListRefresh, updateSessionActivityPhase]); const handleEvent = React.useCallback((event: EventData) => { lastEventTimestampRef.current = Date.now(); @@ -793,7 +901,8 @@ export const useEventStream = () => { } } - if (messageId !== latestAssistantMessageId) break; + const isActiveSession = currentSessionId === sessionId; + if (isActiveSession && messageId !== latestAssistantMessageId) break; if (!stopMarkerPresent && isDesktopRuntimeRef.current) { trackMessage(messageId, 'desktop_completion_without_stop'); @@ -1035,6 +1144,7 @@ export const useEventStream = () => { console.debug('[useEventStream] Connection state:', { isDesktopRuntime: isDesktopRuntimeRef.current, hasUnsubscribe: Boolean(unsubscribeRef.current), + hasActivityStream: Boolean(activityStreamAbortControllerRef.current), currentSessionId: currentSessionIdRef.current, effectiveDirectory, onlineStatus: onlineStatusRef.current, @@ -1073,6 +1183,15 @@ export const useEventStream = () => { } } + if (activityStreamAbortControllerRef.current) { + try { + activityStreamAbortControllerRef.current.abort(); + } catch (error) { + console.warn('[useEventStream] Error during activity stream abort:', error); + } + activityStreamAbortControllerRef.current = null; + } + isCleaningUpRef.current = false; }, []); @@ -1159,7 +1278,143 @@ export const useEventStream = () => { } try { - const sdkUnsub = opencodeClient.subscribeToEvents(handleEvent, onError, onOpen, effectiveDirectory); + const sdkUnsub = opencodeClient.subscribeToEvents( + handleEvent, + onError, + onOpen, + effectiveDirectory, + { scope: 'directory', key: 'events' } + ); + + if (!isDesktopRuntimeRef.current) { + if (activityStreamAbortControllerRef.current) { + activityStreamAbortControllerRef.current.abort(); + } + + const activityAbortController = new AbortController(); + activityStreamAbortControllerRef.current = activityAbortController; + + const parseSseEventBlock = (block: string): EventData | null => { + if (!block) 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 { + const parsed = JSON.parse(payloadText) as unknown; + if (!parsed || typeof parsed !== 'object') { + return null; + } + + const record = parsed as Record; + if (typeof record.type === 'string') { + return record as unknown as EventData; + } + + const nestedPayload = record.payload; + if (nestedPayload && typeof nestedPayload === 'object') { + const nestedRecord = nestedPayload as Record; + if (typeof nestedRecord.type === 'string') { + return nestedRecord as unknown as EventData; + } + } + + return null; + } catch { + return null; + } + }; + + void (async () => { + try { + const candidateEndpoints = ['/api/global/event', '/api/event']; + let response: Response | null = null; + let lastError: unknown = null; + + for (const endpoint of candidateEndpoints) { + try { + const candidateResponse = await fetch(endpoint, { + method: 'GET', + headers: { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }, + signal: activityAbortController.signal, + }); + + if (candidateResponse.ok && candidateResponse.body) { + response = candidateResponse; + if (streamDebugEnabled()) { + console.info('[useEventStream] Activity stream connected:', endpoint); + } + break; + } + + lastError = new Error(`Activity stream failed: ${candidateResponse.status}`); + } catch (error) { + lastError = error; + } + } + + if (!response) { + throw lastError ?? new Error('Activity stream failed'); + } + + const responseBody = response.body; + if (!responseBody) { + throw new Error('Activity stream missing body'); + } + + const reader = responseBody.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (activityAbortController.signal.aborted) break; + if (!value || value.length === 0) continue; + + buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n'); + const blocks = buffer.split('\n\n'); + buffer = blocks.pop() ?? ''; + for (const block of blocks) { + const event = parseSseEventBlock(block); + if (event) { + handleActivityEvent(event); + } + } + } + + const remaining = buffer.trim(); + if (remaining) { + const event = parseSseEventBlock(remaining); + if (event) { + handleActivityEvent(event); + } + } + } catch (error) { + if (!activityAbortController.signal.aborted) { + console.warn('[useEventStream] Activity stream error:', error); + } + } finally { + if (activityStreamAbortControllerRef.current === activityAbortController) { + activityStreamAbortControllerRef.current = null; + } + } + })(); + } const compositeUnsub = () => { try { @@ -1173,6 +1428,10 @@ export const useEventStream = () => { unsubscribeRef.current = compositeUnsub; } else { compositeUnsub(); + if (activityStreamAbortControllerRef.current) { + activityStreamAbortControllerRef.current.abort(); + activityStreamAbortControllerRef.current = null; + } } } catch (subscriptionError) { console.error('[useEventStream] Error during subscription:', subscriptionError); @@ -1186,6 +1445,7 @@ export const useEventStream = () => { resyncMessages, requestSessionMetadataRefresh, handleEvent, + handleActivityEvent, effectiveDirectory, refreshSessionActivityStatus, waitForDesktopBridge, diff --git a/packages/ui/src/lib/opencode/client.ts b/packages/ui/src/lib/opencode/client.ts index 7aa66f20..73e1d645 100644 --- a/packages/ui/src/lib/opencode/client.ts +++ b/packages/ui/src/lib/opencode/client.ts @@ -129,7 +129,7 @@ const getDesktopFilesApi = (): FilesAPI | null => { class OpencodeService { private client: OpencodeClient; private baseUrl: string; - private sseAbortController: AbortController | null = null; + private sseAbortControllers: Map = new Map(); private currentDirectory: string | undefined = undefined; constructor(baseUrl: string = DEFAULT_BASE_URL) { @@ -835,16 +835,18 @@ class OpencodeService { onMessage: (event: { type: string; properties?: Record }) => void, onError?: (error: unknown) => void, onOpen?: () => void, - directoryOverride?: string | null + directoryOverride?: string | null, + options?: { scope?: 'global' | 'directory'; key?: string } ): () => void { - // Stop any existing subscription - if (this.sseAbortController) { - this.sseAbortController.abort(); + const subscriptionKey = options?.key ?? 'default'; + const existingController = this.sseAbortControllers.get(subscriptionKey); + if (existingController) { + existingController.abort(); } // Create new AbortController for this subscription const abortController = new AbortController(); - this.sseAbortController = abortController; + this.sseAbortControllers.set(subscriptionKey, abortController); let lastEventId: string | undefined; @@ -853,10 +855,12 @@ class OpencodeService { // Start async generator in background with reconnect on failure (async () => { const resolvedDirectory = - typeof directoryOverride === 'string' && directoryOverride.trim().length > 0 - ? directoryOverride.trim() - : this.currentDirectory; - console.log('[OpencodeClient] Connecting to SSE with directory:', resolvedDirectory); + options?.scope === 'global' + ? undefined + : typeof directoryOverride === 'string' && directoryOverride.trim().length > 0 + ? directoryOverride.trim() + : this.currentDirectory; + console.log('[OpencodeClient] Connecting to SSE with directory:', resolvedDirectory ?? 'global'); const connect = async (attempt: number): Promise => { try { @@ -939,16 +943,16 @@ class OpencodeService { await connect(0); } finally { console.log('[OpencodeClient] SSE subscription cleanup'); - if (this.sseAbortController === abortController) { - this.sseAbortController = null; + if (this.sseAbortControllers.get(subscriptionKey) === abortController) { + this.sseAbortControllers.delete(subscriptionKey); } } })(); // Return cleanup function return () => { - if (this.sseAbortController === abortController) { - this.sseAbortController = null; + if (this.sseAbortControllers.get(subscriptionKey) === abortController) { + this.sseAbortControllers.delete(subscriptionKey); } abortController.abort(); }; diff --git a/packages/vscode/src/AgentManagerPanelProvider.ts b/packages/vscode/src/AgentManagerPanelProvider.ts index 19851fd7..aa820682 100644 --- a/packages/vscode/src/AgentManagerPanelProvider.ts +++ b/packages/vscode/src/AgentManagerPanelProvider.ts @@ -144,6 +144,7 @@ export class AgentManagerPanelProvider { const { path, headers } = (payload || {}) as { path?: string; headers?: Record }; const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event'; + const shouldInjectActivity = normalizedPath === '/event' || normalizedPath === '/global/event'; if (!apiBaseUrl) { return { @@ -211,11 +212,12 @@ export class AgentManagerPanelProvider { if (!chunk) continue; // Reduce webview message pressure by forwarding complete SSE blocks. - sseBuffer += chunk; + sseBuffer += shouldInjectActivity ? chunk.replace(/\r\n/g, '\n') : chunk; const blocks = sseBuffer.split('\n\n'); sseBuffer = blocks.pop() ?? ''; if (blocks.length > 0) { - const joined = blocks.map((block) => `${block}\n\n`).join(''); + const outboundBlocks = shouldInjectActivity ? expandSseBlocksWithActivity(blocks) : blocks; + const joined = outboundBlocks.map((block: string) => `${block}\n\n`).join(''); this._panel?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: joined }); } } @@ -223,10 +225,16 @@ export class AgentManagerPanelProvider { const tail = decoder.decode(); if (tail) { - sseBuffer += tail; + sseBuffer += shouldInjectActivity ? tail.replace(/\r\n/g, '\n') : tail; } if (sseBuffer) { - this._panel?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: sseBuffer }); + if (shouldInjectActivity) { + const outboundBlocks = expandSseBlocksWithActivity([sseBuffer]); + const joined = outboundBlocks.map((block: string) => `${block}\n\n`).join(''); + this._panel?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: joined }); + } else { + this._panel?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: sseBuffer }); + } } } finally { try { @@ -286,3 +294,116 @@ export class AgentManagerPanelProvider { }); } } + +type SessionActivityPhase = 'idle' | 'busy' | 'cooldown'; + +type SessionActivity = { + sessionId: string; + phase: SessionActivityPhase; +}; + +const parseSseDataPayload = (block: string): Record | null => { + if (!block) { + 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 { + const parsed = JSON.parse(payloadText) as unknown; + if (parsed && typeof parsed === 'object') { + const record = parsed as Record; + const nestedPayload = record.payload; + if (nestedPayload && typeof nestedPayload === 'object') { + return nestedPayload as Record; + } + return record; + } + return null; + } catch { + return null; + } +}; + +const deriveSessionActivity = (payload: Record | null): SessionActivity | null => { + if (!payload || typeof payload !== 'object') { + return null; + } + + const type = payload.type; + const properties = payload.properties as Record | undefined; + + if (type === 'session.status') { + const status = properties?.status as Record | undefined; + const sessionId = properties?.sessionID ?? properties?.sessionId; + const statusType = status?.type; + + if (typeof sessionId === 'string' && sessionId.length > 0 && typeof statusType === 'string') { + const phase: SessionActivityPhase = statusType === 'busy' || statusType === 'retry' ? 'busy' : 'idle'; + return { sessionId, phase }; + } + } + + if (type === 'message.updated') { + const info = properties?.info as Record | undefined; + const sessionId = info?.sessionID ?? info?.sessionId ?? properties?.sessionID ?? properties?.sessionId; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (type === 'message.part.updated') { + const info = properties?.info as Record | undefined; + const sessionId = info?.sessionID ?? info?.sessionId ?? properties?.sessionID ?? properties?.sessionId; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (type === 'session.idle') { + const sessionId = properties?.sessionID ?? properties?.sessionId; + if (typeof sessionId === 'string' && sessionId.length > 0) { + return { sessionId, phase: 'idle' }; + } + } + + return null; +}; + +const buildActivityEventBlock = (activity: SessionActivity): string => { + return `data: ${JSON.stringify({ + type: 'openchamber:session-activity', + properties: { + sessionId: activity.sessionId, + phase: activity.phase, + }, + })}`; +}; + +const expandSseBlocksWithActivity = (blocks: string[]): string[] => { + const expanded: string[] = []; + for (const block of blocks) { + expanded.push(block); + const activity = deriveSessionActivity(parseSseDataPayload(block)); + if (activity) { + expanded.push(buildActivityEventBlock(activity)); + } + } + return expanded; +}; diff --git a/packages/vscode/src/ChatViewProvider.ts b/packages/vscode/src/ChatViewProvider.ts index 47c825de..0f28d883 100644 --- a/packages/vscode/src/ChatViewProvider.ts +++ b/packages/vscode/src/ChatViewProvider.ts @@ -177,6 +177,7 @@ export class ChatViewProvider implements vscode.WebviewViewProvider { const { path, headers } = (payload || {}) as { path?: string; headers?: Record }; const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event'; + const shouldInjectActivity = normalizedPath === '/event' || normalizedPath === '/global/event'; if (!apiBaseUrl) { return { @@ -246,11 +247,12 @@ export class ChatViewProvider implements vscode.WebviewViewProvider { // Reduce webview message pressure by forwarding complete SSE blocks. // The SDK SSE parser is block-based (\n\n delimited) and can consume // partial chunks, but VS Code's postMessage channel can be a bottleneck. - sseBuffer += chunk; + sseBuffer += shouldInjectActivity ? chunk.replace(/\r\n/g, '\n') : chunk; const blocks = sseBuffer.split('\n\n'); sseBuffer = blocks.pop() ?? ''; if (blocks.length > 0) { - const joined = blocks.map((block) => `${block}\n\n`).join(''); + const outboundBlocks = shouldInjectActivity ? expandSseBlocksWithActivity(blocks) : blocks; + const joined = outboundBlocks.map((block: string) => `${block}\n\n`).join(''); this._view?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: joined }); } } @@ -258,10 +260,16 @@ export class ChatViewProvider implements vscode.WebviewViewProvider { const tail = decoder.decode(); if (tail) { - sseBuffer += tail; + sseBuffer += shouldInjectActivity ? tail.replace(/\r\n/g, '\n') : tail; } if (sseBuffer) { - this._view?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: sseBuffer }); + if (shouldInjectActivity) { + const outboundBlocks = expandSseBlocksWithActivity([sseBuffer]); + const joined = outboundBlocks.map((block: string) => `${block}\n\n`).join(''); + this._view?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: joined }); + } else { + this._view?.webview.postMessage({ type: 'api:sse:chunk', streamId, chunk: sseBuffer }); + } } } finally { try { @@ -322,3 +330,116 @@ export class ChatViewProvider implements vscode.WebviewViewProvider { }); } } + +type SessionActivityPhase = 'idle' | 'busy' | 'cooldown'; + +type SessionActivity = { + sessionId: string; + phase: SessionActivityPhase; +}; + +const parseSseDataPayload = (block: string): Record | null => { + if (!block) { + 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 { + const parsed = JSON.parse(payloadText) as unknown; + if (parsed && typeof parsed === 'object') { + const record = parsed as Record; + const nestedPayload = record.payload; + if (nestedPayload && typeof nestedPayload === 'object') { + return nestedPayload as Record; + } + return record; + } + return null; + } catch { + return null; + } +}; + +const deriveSessionActivity = (payload: Record | null): SessionActivity | null => { + if (!payload || typeof payload !== 'object') { + return null; + } + + const type = payload.type; + const properties = payload.properties as Record | undefined; + + if (type === 'session.status') { + const status = properties?.status as Record | undefined; + const sessionId = properties?.sessionID ?? properties?.sessionId; + const statusType = status?.type; + + if (typeof sessionId === 'string' && sessionId.length > 0 && typeof statusType === 'string') { + const phase: SessionActivityPhase = statusType === 'busy' || statusType === 'retry' ? 'busy' : 'idle'; + return { sessionId, phase }; + } + } + + if (type === 'message.updated') { + const info = properties?.info as Record | undefined; + const sessionId = info?.sessionID ?? info?.sessionId ?? properties?.sessionID ?? properties?.sessionId; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (type === 'message.part.updated') { + const info = properties?.info as Record | undefined; + const sessionId = info?.sessionID ?? info?.sessionId ?? properties?.sessionID ?? properties?.sessionId; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (type === 'session.idle') { + const sessionId = properties?.sessionID ?? properties?.sessionId; + if (typeof sessionId === 'string' && sessionId.length > 0) { + return { sessionId, phase: 'idle' }; + } + } + + return null; +}; + +const buildActivityEventBlock = (activity: SessionActivity): string => { + return `data: ${JSON.stringify({ + type: 'openchamber:session-activity', + properties: { + sessionId: activity.sessionId, + phase: activity.phase, + }, + })}`; +}; + +const expandSseBlocksWithActivity = (blocks: string[]): string[] => { + const expanded: string[] = []; + for (const block of blocks) { + expanded.push(block); + const activity = deriveSessionActivity(parseSseDataPayload(block)); + if (activity) { + expanded.push(buildActivityEventBlock(activity)); + } + } + return expanded; +}; diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 09fcbe5a..6873a4ea 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -937,7 +937,16 @@ function parseSseDataPayload(block) { } try { - return JSON.parse(payloadText); + const parsed = JSON.parse(payloadText); + if ( + parsed && + typeof parsed === 'object' && + typeof parsed.payload === 'object' && + parsed.payload !== null + ) { + return parsed.payload; + } + return parsed; } catch { return null; } @@ -950,7 +959,7 @@ function deriveSessionActivity(payload) { if (payload.type === 'session.status') { const status = payload.properties?.status; - const sessionId = payload.properties?.sessionID; + const sessionId = payload.properties?.sessionID ?? payload.properties?.sessionId; const statusType = status?.type; if (typeof sessionId === 'string' && sessionId.length > 0 && typeof statusType === 'string') { @@ -961,7 +970,17 @@ function deriveSessionActivity(payload) { if (payload.type === 'message.updated') { const info = payload.properties?.info; - const sessionId = info?.sessionID; + const sessionId = info?.sessionID ?? info?.sessionId ?? payload.properties?.sessionID ?? payload.properties?.sessionId; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (payload.type === 'message.part.updated') { + const info = payload.properties?.info; + const sessionId = info?.sessionID ?? info?.sessionId ?? payload.properties?.sessionID ?? payload.properties?.sessionId; const role = info?.role; const finish = info?.finish; if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { @@ -970,7 +989,7 @@ function deriveSessionActivity(payload) { } if (payload.type === 'session.idle') { - const sessionId = payload.properties?.sessionID; + const sessionId = payload.properties?.sessionID ?? payload.properties?.sessionId; if (typeof sessionId === 'string' && sessionId.length > 0) { return { sessionId, phase: 'idle' }; } @@ -2072,11 +2091,119 @@ async function main(options = {}) { } }); - app.get('/api/event', async (req, res) => { - if (!openCodePort) { + app.get('/api/global/event', async (req, res) => { + if (!openCodeApiPrefixDetected) { + try { + await detectOpenCodeApiPrefix(); + } catch { + // ignore detection failures + } + } + + let targetUrl; + try { + const prefix = openCodeApiPrefixDetected ? openCodeApiPrefix : ''; + targetUrl = new URL(buildOpenCodeUrl('/global/event', prefix)); + } catch (error) { return res.status(503).json({ error: 'OpenCode service unavailable' }); } + 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/event', async (req, res) => { if (!openCodeApiPrefixDetected) { try { await detectOpenCodeApiPrefix();