diff --git a/packages/ui/src/components/chat/ChatInput.tsx b/packages/ui/src/components/chat/ChatInput.tsx index d2302d6e..2f04f958 100644 --- a/packages/ui/src/components/chat/ChatInput.tsx +++ b/packages/ui/src/components/chat/ChatInput.tsx @@ -2880,6 +2880,8 @@ const ChatInputComponent: React.FC = ({ onOpenSettings, scrollTo }; }, [fetchBranches, runtimeGit, selectedDraftProject, selectedDraftProjectBranches?.all, selectedDraftProjectPath, showDraftTargetSelectors]); + const selectedDraftProjectCurrentBranch = selectedDraftProjectBranches?.current?.trim() ?? ''; + const projectRootBranchOption = React.useMemo(() => { if (!selectedDraftProject) { return null; @@ -2888,15 +2890,14 @@ const ChatInputComponent: React.FC = ({ onOpenSettings, scrollTo if (!value) { return null; } - const projectRootBranch = selectedDraftProjectBranches?.current?.trim() ?? ''; - if (!projectRootBranch) { + if (!selectedDraftProjectCurrentBranch) { return null; } return { value, - label: projectRootBranch, + label: selectedDraftProjectCurrentBranch, }; - }, [selectedDraftProject, selectedDraftProjectBranches]); + }, [selectedDraftProject, selectedDraftProjectCurrentBranch]); const worktreeBranchOptions = React.useMemo(() => { if (!selectedDraftProject) { @@ -2914,11 +2915,11 @@ const ChatInputComponent: React.FC = ({ onOpenSettings, scrollTo return buildSessionTargetOptions({ projectRoot: normalizePath(selectedDraftProject.path) ?? '', - rootBranch: selectedDraftProjectBranches?.current?.trim() ?? '', + rootBranch: selectedDraftProjectCurrentBranch, worktrees, pendingBootstrapDirectory: newSessionDraft?.bootstrapPendingDirectory ?? null, }); - }, [availableWorktreesByProject, newSessionDraft?.bootstrapPendingDirectory, selectedDraftProject, selectedDraftProjectBranches?.current, selectedDraftProjectPath]); + }, [availableWorktreesByProject, newSessionDraft?.bootstrapPendingDirectory, selectedDraftProject, selectedDraftProjectCurrentBranch, selectedDraftProjectPath]); const selectedDraftDirectory = React.useMemo( () => normalizePath(newSessionDraft?.bootstrapPendingDirectory ?? null) diff --git a/packages/ui/src/components/sections/openchamber/OpenChamberPage.tsx b/packages/ui/src/components/sections/openchamber/OpenChamberPage.tsx index d233192e..8fa8c43c 100644 --- a/packages/ui/src/components/sections/openchamber/OpenChamberPage.tsx +++ b/packages/ui/src/components/sections/openchamber/OpenChamberPage.tsx @@ -122,7 +122,7 @@ const VisualSectionContent: React.FC = () => { // Chat section: User message rendering, Diff layout, Mobile status bar, Show reasoning traces, Queue mode, Persist draft const ChatSectionContent: React.FC = () => { - return ; + return ; }; // Sessions section: Default model & agent, Session retention diff --git a/packages/ui/src/components/sections/openchamber/OpenChamberVisualSettings.tsx b/packages/ui/src/components/sections/openchamber/OpenChamberVisualSettings.tsx index b3ce6e51..cb460628 100644 --- a/packages/ui/src/components/sections/openchamber/OpenChamberVisualSettings.tsx +++ b/packages/ui/src/components/sections/openchamber/OpenChamberVisualSettings.tsx @@ -23,6 +23,7 @@ import { isDesktopShell, isVSCodeRuntime, isWebRuntime } from '@/lib/desktop'; import { useDeviceInfo } from '@/lib/device'; import { usePwaDetection } from '@/hooks/usePwaDetection'; import { updateDesktopSettings } from '@/lib/persistence'; +import { useConfigStore } from '@/stores/useConfigStore'; import { setDirectoryShowHidden, useDirectoryShowHidden, @@ -126,6 +127,24 @@ const CHAT_RENDER_MODE_OPTIONS: Option<'sorted' | 'live'>[] = [ }, ]; +const MESSAGE_STREAM_TRANSPORT_OPTIONS: Option<'auto' | 'ws' | 'sse'>[] = [ + { + id: 'auto', + label: 'Auto', + description: 'Prefer WebSocket and fall back to SSE if needed.', + }, + { + id: 'ws', + label: 'WebSocket', + description: 'Use WebSocket for message streaming.', + }, + { + id: 'sse', + label: 'SSE', + description: 'Use Server-Sent Events for message streaming.', + }, +]; + const ACTIVITY_RENDER_MODE_OPTIONS: Option<'collapsed' | 'summary'>[] = [ { id: 'collapsed', @@ -177,7 +196,7 @@ const normalizeUserMessageRenderingMode = (mode: unknown): 'markdown' | 'plain' return mode === 'markdown' ? 'markdown' : 'plain'; }; -export type VisibleSetting = 'theme' | 'pwaInstallName' | 'timeFormat' | 'weekStart' | 'fontSize' | 'terminalFontSize' | 'spacing' | 'inputBarOffset' | 'mermaidRendering' | 'userMessageRendering' | 'chatRenderMode' | 'activityRenderMode' | 'stickyUserHeader' | 'diffLayout' | 'mobileStatusBar' | 'dotfiles' | 'reasoning' | 'showToolFileIcons' | 'expandedTools' | 'queueMode' | 'terminalQuickKeys' | 'persistDraft' | 'inputSpellcheck' | 'reportUsage'; +export type VisibleSetting = 'theme' | 'pwaInstallName' | 'timeFormat' | 'weekStart' | 'fontSize' | 'terminalFontSize' | 'spacing' | 'inputBarOffset' | 'mermaidRendering' | 'userMessageRendering' | 'chatRenderMode' | 'messageTransport' | 'activityRenderMode' | 'stickyUserHeader' | 'diffLayout' | 'mobileStatusBar' | 'dotfiles' | 'reasoning' | 'showToolFileIcons' | 'expandedTools' | 'queueMode' | 'terminalQuickKeys' | 'persistDraft' | 'inputSpellcheck' | 'reportUsage'; interface OpenChamberVisualSettingsProps { /** Which settings to show. If undefined, shows all. */ @@ -233,6 +252,8 @@ export const OpenChamberVisualSettings: React.FC const setWeekStartPreference = useUIStore(state => state.setWeekStartPreference); const showMobileSessionStatusBar = useUIStore(state => state.showMobileSessionStatusBar); const setShowMobileSessionStatusBar = useUIStore(state => state.setShowMobileSessionStatusBar); + const messageStreamTransport = useConfigStore((state) => state.settingsMessageStreamTransport); + const setMessageStreamTransport = useConfigStore((state) => state.setSettingsMessageStreamTransport); const isSettingsDialogOpen = useUIStore(state => state.isSettingsDialogOpen); const { themeMode, @@ -323,6 +344,11 @@ export const OpenChamberVisualSettings: React.FC void updateDesktopSettings({ chatRenderMode: mode }); }, [setChatRenderMode]); + const handleMessageStreamTransportChange = React.useCallback((mode: 'auto' | 'ws' | 'sse') => { + setMessageStreamTransport(mode); + void updateDesktopSettings({ messageStreamTransport: mode }); + }, [setMessageStreamTransport]); + const handleActivityRenderModeChange = React.useCallback((mode: 'collapsed' | 'summary') => { setActivityRenderMode(mode); void updateDesktopSettings({ activityRenderMode: mode }); @@ -399,6 +425,7 @@ export const OpenChamberVisualSettings: React.FC const hasBehaviorSettings = shouldShow('mermaidRendering') || shouldShow('userMessageRendering') || shouldShow('chatRenderMode') + || shouldShow('messageTransport') || (shouldShow('activityRenderMode') && chatRenderMode === 'sorted') || shouldShow('stickyUserHeader') || shouldShow('diffLayout') @@ -848,7 +875,7 @@ export const OpenChamberVisualSettings: React.FC - {(shouldShow('userMessageRendering') || shouldShow('mermaidRendering') || shouldShow('chatRenderMode') || (shouldShow('activityRenderMode') && chatRenderMode === 'sorted') || (shouldShow('diffLayout') && !isVSCode)) && ( + {(shouldShow('userMessageRendering') || shouldShow('mermaidRendering') || shouldShow('chatRenderMode') || shouldShow('messageTransport') || (shouldShow('activityRenderMode') && chatRenderMode === 'sorted') || (shouldShow('diffLayout') && !isVSCode)) && (
{shouldShow('chatRenderMode') && (
@@ -934,6 +961,35 @@ export const OpenChamberVisualSettings: React.FC
)} + {shouldShow('messageTransport') && ( +
+

Message Stream Transport

+
+
+ {MESSAGE_STREAM_TRANSPORT_OPTIONS.map((option) => ( + + ))} +
+ + {MESSAGE_STREAM_TRANSPORT_OPTIONS.find((option) => option.id === messageStreamTransport)?.description} + +
+
+ )} + {shouldShow('activityRenderMode') && chatRenderMode === 'sorted' && (

Activity Default

diff --git a/packages/ui/src/components/session/sidebar/hooks/useDirectoryStatusProbe.ts b/packages/ui/src/components/session/sidebar/hooks/useDirectoryStatusProbe.ts index 9e6ebf27..29d3e547 100644 --- a/packages/ui/src/components/session/sidebar/hooks/useDirectoryStatusProbe.ts +++ b/packages/ui/src/components/session/sidebar/hooks/useDirectoryStatusProbe.ts @@ -44,18 +44,29 @@ async function probeDirectory(directory: string): Promise try { await opencodeClient.listLocalDirectory(directory); return 'exists'; - } catch { + } catch (error) { const looksLikeSdkWorktree = directory.includes('/opencode/worktree/') || directory.includes('/.opencode/data/worktree/') || directory.includes('/.local/share/opencode/worktree/'); - if (looksLikeSdkWorktree) { - const ok = await opencodeClient.probeDirectory(directory).catch(() => false); - if (ok) return 'exists'; + const reachable = await opencodeClient.probeDirectory(directory).catch(() => false); + if (reachable) { + return 'exists'; } - return 'missing'; + const message = error instanceof Error ? error.message.toLowerCase() : ''; + const definitelyMissing = + message.includes('enoent') || + message.includes('not found') || + message.includes('does not exist') || + message.includes('no such file'); + + if (definitelyMissing || looksLikeSdkWorktree) { + return 'missing'; + } + + return 'unknown'; } } @@ -75,13 +86,17 @@ export const useDirectoryStatusProbe = ({ React.useEffect(() => { const directories = new Set(); + const normalizedProjectRoots = new Set(); sortedSessions.forEach((session) => { const dir = normalizePath((session as Session & { directory?: string | null }).directory ?? null); if (dir) directories.add(dir); }); projects.forEach((project) => { const normalized = normalizePath(project.path); - if (normalized) directories.add(normalized); + if (normalized) { + directories.add(normalized); + normalizedProjectRoots.add(normalized); + } }); const now = Date.now(); @@ -90,12 +105,25 @@ export const useDirectoryStatusProbe = ({ const preseeded = new Map(); for (const directory of directories) { + const isProjectRoot = normalizedProjectRoots.has(directory); const known = directoryStatusRef.current.get(directory); - if (known && known !== 'unknown') continue; + if (known === 'exists') continue; + + const cachedAt = missingCache[directory]; + if (known === 'missing') { + if (isProjectRoot) { + toProbe.push(directory); + continue; + } + if (cachedAt && now - cachedAt < MISSING_REPROBE_MS) { + continue; + } + toProbe.push(directory); + continue; + } // Use cached "missing" status if fresh enough — skip the HTTP probe - const cachedAt = missingCache[directory]; - if (cachedAt && now - cachedAt < MISSING_REPROBE_MS) { + if (!isProjectRoot && cachedAt && now - cachedAt < MISSING_REPROBE_MS) { preseeded.set(directory, 'missing'); continue; } diff --git a/packages/ui/src/lib/api/types.ts b/packages/ui/src/lib/api/types.ts index 612de64a..0876f509 100644 --- a/packages/ui/src/lib/api/types.ts +++ b/packages/ui/src/lib/api/types.ts @@ -569,6 +569,7 @@ export interface SettingsPayload { showExpandedBashTools?: boolean; showExpandedEditTools?: boolean; chatRenderMode?: 'sorted' | 'live'; + messageStreamTransport?: 'auto' | 'ws' | 'sse'; activityRenderMode?: 'collapsed' | 'summary'; mermaidRenderingMode?: 'svg' | 'ascii'; fontSize?: number; diff --git a/packages/ui/src/lib/desktop.ts b/packages/ui/src/lib/desktop.ts index 12d5ca08..3836281e 100644 --- a/packages/ui/src/lib/desktop.ts +++ b/packages/ui/src/lib/desktop.ts @@ -125,6 +125,7 @@ export type DesktopSettings = { timeFormatPreference?: 'auto' | '12h' | '24h'; weekStartPreference?: 'auto' | 'sunday' | 'monday'; chatRenderMode?: 'sorted' | 'live'; + messageStreamTransport?: 'auto' | 'ws' | 'sse'; activityRenderMode?: 'collapsed' | 'summary'; mermaidRenderingMode?: 'svg' | 'ascii'; userMessageRenderingMode?: 'markdown' | 'plain'; diff --git a/packages/ui/src/lib/persistence.ts b/packages/ui/src/lib/persistence.ts index 9c4c6f83..2d06acaa 100644 --- a/packages/ui/src/lib/persistence.ts +++ b/packages/ui/src/lib/persistence.ts @@ -297,6 +297,9 @@ const getRuntimeSettingsAPI = () => getRegisteredRuntimeAPIs()?.settings ?? null const applyDesktopUiPreferences = (settings: DesktopSettings) => { const store = useUIStore.getState(); + const configStore = typeof window !== 'undefined' + ? window.__zustand_config_store__?.getState?.() ?? null + : null; const queueStore = useMessageQueueStore.getState(); if (typeof settings.showReasoningTraces === 'boolean' && settings.showReasoningTraces !== store.showReasoningTraces) { @@ -407,6 +410,12 @@ const applyDesktopUiPreferences = (settings: DesktopSettings) => { store.setUserMessageRenderingMode(settings.userMessageRenderingMode); } } + if (typeof settings.messageStreamTransport === 'string' + && (settings.messageStreamTransport === 'auto' || settings.messageStreamTransport === 'ws' || settings.messageStreamTransport === 'sse')) { + if (configStore && settings.messageStreamTransport !== configStore.settingsMessageStreamTransport) { + configStore.setSettingsMessageStreamTransport(settings.messageStreamTransport); + } + } if (typeof settings.stickyUserHeader === 'boolean' && settings.stickyUserHeader !== store.stickyUserHeader) { store.setStickyUserHeader(settings.stickyUserHeader); } @@ -804,6 +813,10 @@ const sanitizeWebSettings = (payload: unknown): DesktopSettings | null => { && (candidate.chatRenderMode === 'sorted' || candidate.chatRenderMode === 'live')) { result.chatRenderMode = candidate.chatRenderMode; } + if (typeof candidate.messageStreamTransport === 'string' + && (candidate.messageStreamTransport === 'auto' || candidate.messageStreamTransport === 'ws' || candidate.messageStreamTransport === 'sse')) { + result.messageStreamTransport = candidate.messageStreamTransport; + } if (typeof candidate.activityRenderMode === 'string' && (candidate.activityRenderMode === 'collapsed' || candidate.activityRenderMode === 'summary')) { result.activityRenderMode = candidate.activityRenderMode; diff --git a/packages/ui/src/stores/useConfigStore.ts b/packages/ui/src/stores/useConfigStore.ts index b4e939eb..9226e888 100644 --- a/packages/ui/src/stores/useConfigStore.ts +++ b/packages/ui/src/stores/useConfigStore.ts @@ -29,6 +29,7 @@ interface OpenChamberDefaults { autoCreateWorktree?: boolean; gitmojiEnabled?: boolean; zenModel?: string; + messageStreamTransport?: 'auto' | 'ws' | 'sse'; } const fetchOpenChamberDefaults = async (): Promise => { @@ -45,6 +46,10 @@ const fetchOpenChamberDefaults = async (): Promise => { const defaultAgent = typeof data?.defaultAgent === 'string' ? data.defaultAgent.trim() : ''; const gitmojiEnabled = typeof data?.gitmojiEnabled === 'boolean' ? data.gitmojiEnabled : undefined; const zenModel = typeof data?.zenModel === 'string' ? data.zenModel.trim() : ''; + const messageStreamTransport = + data?.messageStreamTransport === 'ws' || data?.messageStreamTransport === 'sse' || data?.messageStreamTransport === 'auto' + ? data.messageStreamTransport + : undefined; return { defaultModel: defaultModel.length > 0 ? defaultModel : undefined, @@ -53,6 +58,7 @@ const fetchOpenChamberDefaults = async (): Promise => { autoCreateWorktree: typeof data?.autoCreateWorktree === 'boolean' ? data.autoCreateWorktree : undefined, gitmojiEnabled, zenModel: zenModel.length > 0 ? zenModel : undefined, + messageStreamTransport, }; } } catch { @@ -74,6 +80,10 @@ const fetchOpenChamberDefaults = async (): Promise => { const defaultAgent = typeof data?.defaultAgent === 'string' ? data.defaultAgent.trim() : ''; const gitmojiEnabled = typeof data?.gitmojiEnabled === 'boolean' ? data.gitmojiEnabled : undefined; const zenModel = typeof data?.zenModel === 'string' ? data.zenModel.trim() : ''; + const messageStreamTransport = + data?.messageStreamTransport === 'ws' || data?.messageStreamTransport === 'sse' || data?.messageStreamTransport === 'auto' + ? data.messageStreamTransport + : undefined; return { defaultModel: defaultModel.length > 0 ? defaultModel : undefined, @@ -82,6 +92,7 @@ const fetchOpenChamberDefaults = async (): Promise => { autoCreateWorktree: typeof data?.autoCreateWorktree === 'boolean' ? data.autoCreateWorktree : undefined, gitmojiEnabled, zenModel: zenModel.length > 0 ? zenModel : undefined, + messageStreamTransport, }; } catch { return {}; @@ -466,6 +477,7 @@ interface ConfigStore { settingsAutoCreateWorktree: boolean; settingsGitmojiEnabled: boolean; settingsZenModel: string | undefined; + settingsMessageStreamTransport: 'auto' | 'ws' | 'sse'; // Voice provider preference ('browser', 'openai', 'openai-compatible', or 'say' for macOS) voiceProvider: 'browser' | 'openai' | 'openai-compatible' | 'say'; setVoiceProvider: (provider: 'browser' | 'openai' | 'openai-compatible' | 'say') => void; @@ -534,6 +546,7 @@ interface ConfigStore { setSettingsAutoCreateWorktree: (enabled: boolean) => void; setSettingsGitmojiEnabled: (enabled: boolean) => void; setSettingsZenModel: (model: string | undefined) => void; + setSettingsMessageStreamTransport: (transport: 'auto' | 'ws' | 'sse') => void; getResolvedGitGenerationModel: () => { providerId: string; modelId: string } | null; saveAgentModelSelection: (agentName: string, providerId: string, modelId: string) => void; getAgentModelSelection: (agentName: string) => { providerId: string; modelId: string } | null; @@ -583,6 +596,7 @@ export const useConfigStore = create()( settingsAutoCreateWorktree: false, settingsGitmojiEnabled: false, settingsZenModel: undefined, + settingsMessageStreamTransport: 'auto', // Voice provider preference - load from localStorage or default to 'browser' voiceProvider: (() => { if (typeof window !== 'undefined') { @@ -1256,6 +1270,7 @@ export const useConfigStore = create()( settingsAutoCreateWorktree: openChamberDefaults.autoCreateWorktree ?? false, settingsGitmojiEnabled: openChamberDefaults.gitmojiEnabled ?? false, settingsZenModel: resolvedZenModel, + settingsMessageStreamTransport: openChamberDefaults.messageStreamTransport ?? state.settingsMessageStreamTransport ?? 'auto', directoryScoped: { ...state.directoryScoped, [directoryKey]: nextSnapshot, @@ -1687,6 +1702,10 @@ export const useConfigStore = create()( set({ settingsZenModel: model }); }, + setSettingsMessageStreamTransport: (transport: 'auto' | 'ws' | 'sse') => { + set({ settingsMessageStreamTransport: transport }); + }, + getResolvedGitGenerationModel: () => { const state = get(); return resolveGitGenerationModelSelection({ @@ -1983,6 +2002,7 @@ export const useConfigStore = create()( settingsAutoCreateWorktree: state.settingsAutoCreateWorktree, settingsGitmojiEnabled: state.settingsGitmojiEnabled, settingsZenModel: state.settingsZenModel, + settingsMessageStreamTransport: state.settingsMessageStreamTransport, speechRate: state.speechRate, speechPitch: state.speechPitch, speechVolume: state.speechVolume, diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js index afabd19a..5914212e 100644 --- a/packages/ui/src/sync/__tests__/event-pipeline.test.js +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -3,6 +3,7 @@ import { createEventPipeline } from '../event-pipeline'; const originalDocument = globalThis.document; const originalWindow = globalThis.window; +const originalWebSocket = globalThis.WebSocket; function installDomStubs() { globalThis.document = { @@ -12,14 +13,52 @@ function installDomStubs() { }; globalThis.window = { + location: { + href: 'http://127.0.0.1:3000/', + origin: 'http://127.0.0.1:3000', + }, addEventListener() {}, removeEventListener() {}, }; } +class FakeWebSocket { + static instances = []; + + constructor(url) { + this.url = url; + this.readyState = 0; + this.onopen = null; + this.onmessage = null; + this.onerror = null; + this.onclose = null; + FakeWebSocket.instances.push(this); + } + + close() { + this.readyState = 3; + } + + emitOpen() { + this.readyState = 1; + this.onopen?.(); + } + + emitMessage(payload) { + this.onmessage?.({ data: JSON.stringify(payload) }); + } + + emitClose() { + this.readyState = 3; + this.onclose?.(); + } +} + afterEach(() => { globalThis.document = originalDocument; globalThis.window = originalWindow; + globalThis.WebSocket = originalWebSocket; + FakeWebSocket.instances = []; }); function createSdkWithSingleEvent(event, hold) { @@ -473,6 +512,160 @@ describe('createEventPipeline', () => { expect(received[0].payload.type).toBe('message.part.updated'); expect(received[0].payload.properties.part.text).toBe('next'); }); + + it('consumes websocket message stream frames when transport is ws', async () => { + installDomStubs(); + globalThis.WebSocket = FakeWebSocket; + + const received = []; + const sdk = { + global: { + event: async () => { + throw new Error('SSE should not be used in ws mode'); + }, + }, + }; + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + transport: 'ws', + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + resolve(); + }, + }); + }); + + await Promise.resolve(); + + const socket = FakeWebSocket.instances[0]; + expect(socket?.url).toContain('/api/global/event/ws'); + + socket.emitOpen(); + socket.emitMessage({ type: 'ready', scope: 'global' }); + socket.emitMessage({ + type: 'event', + eventId: 'evt-1', + directory: '/tmp/project', + payload: { + type: 'session.status', + properties: { + sessionID: 'session-1', + }, + }, + }); + + await delivered; + + expect(received).toEqual([ + { + directory: '/tmp/project', + payload: { + type: 'session.status', + properties: { + sessionID: 'session-1', + }, + }, + }, + ]); + }); + + it('falls back to SSE when websocket closes before ready in auto mode', async () => { + installDomStubs(); + globalThis.WebSocket = FakeWebSocket; + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + payload: { + type: 'server.connected', + properties: {}, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + transport: 'auto', + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await Promise.resolve(); + const socket = FakeWebSocket.instances[0]; + socket.emitClose(); + + await delivered; + + expect(received).toEqual([ + { + directory: 'global', + payload: { + type: 'server.connected', + properties: {}, + }, + }, + ]); + }); + + it('falls back to SSE when websocket does not become ready in auto mode', async () => { + installDomStubs(); + globalThis.WebSocket = FakeWebSocket; + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + payload: { + type: 'server.connected', + properties: {}, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + transport: 'auto', + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await Promise.resolve(); + const socket = FakeWebSocket.instances[0]; + socket.emitOpen(); + + await new Promise((resolve) => setTimeout(resolve, 2300)); + await delivered; + + expect(received).toEqual([ + { + directory: 'global', + payload: { + type: 'server.connected', + properties: {}, + }, + }, + ]); + }); }); // --------------------------------------------------------------------------- diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index e91676b5..6ebb2e41 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -1,5 +1,5 @@ /** - * Event Pipeline — SSE connection, event coalescing, and batched flush. + * Event Pipeline — transport connection, event coalescing, and batched flush. * * Plain closure API: * const { cleanup } = createEventPipeline({ sdk, onEvent }) @@ -9,12 +9,9 @@ */ import type { Event, OpencodeClient } from "@opencode-ai/sdk/v2/client" +import { opencodeClient } from "@/lib/opencode/client" import { syncDebug } from "./debug" -// --------------------------------------------------------------------------- -// Types -// --------------------------------------------------------------------------- - export type QueuedEvent = { directory: string payload: Event @@ -22,25 +19,30 @@ export type QueuedEvent = { export type FlushHandler = (events: QueuedEvent[]) => void -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- - const FLUSH_FRAME_MS = 16 const STREAM_YIELD_MS = 8 const RECONNECT_DELAY_MS = 250 const HEARTBEAT_TIMEOUT_MS = 15_000 - -// --------------------------------------------------------------------------- -// Pipeline factory -// --------------------------------------------------------------------------- +const WS_FALLBACK_WINDOW_MS = 60_000 +const WS_READY_TIMEOUT_MS = 2_000 +const ABSOLUTE_URL_PATTERN = /^[a-zA-Z][a-zA-Z\d+\-.]*:\/\// export type EventPipelineInput = { sdk: OpencodeClient onEvent: (directory: string, payload: Event) => void routeDirectory?: (directory: string, payload: Event) => string - /** Called after SSE reconnects (visibility restore or heartbeat timeout). */ + /** Called after stream reconnects (visibility restore or heartbeat timeout). */ onReconnect?: () => void + transport?: "auto" | "ws" | "sse" +} + +type MessageStreamWsFrame = { + type: "ready" | "event" | "error" + payload?: unknown + eventId?: string + directory?: string + message?: string + scope?: "global" | "directory" } const normalizeEventType = (payload: Event): Event => { @@ -79,9 +81,57 @@ function resolveEventDirectory(event: unknown, payload: Event): string { return propertyDirectory && propertyDirectory.length > 0 ? propertyDirectory : "global" } -// Per-directory queue state. Each directory owns an independent flush timer -// so a busy directory's delta storm cannot block another directory's events -// from reaching the UI (head-of-line blocking across sessions). +function resolveEventPayload(payload: unknown): Event | null { + if (!payload || typeof payload !== "object") { + return null + } + + const record = payload as { type?: unknown; payload?: unknown } + if (typeof record.type === "string") { + return payload as Event + } + + if (record.payload && typeof record.payload === "object" && typeof (record.payload as { type?: unknown }).type === "string") { + return record.payload as Event + } + + return null +} + +function resolveAbsoluteUrl(candidate: string): string { + const normalized = typeof candidate === "string" && candidate.trim().length > 0 ? candidate.trim() : "/api" + if (ABSOLUTE_URL_PATTERN.test(normalized)) { + return normalized + } + + if (typeof window === "undefined") { + return normalized + } + + const baseReference = window.location?.href || window.location?.origin + if (!baseReference) { + return normalized + } + + return new URL(normalized, baseReference).toString() +} + +function toWebSocketUrl(candidate: string): string { + const url = new URL(resolveAbsoluteUrl(candidate)) + url.protocol = url.protocol === "https:" ? "wss:" : "ws:" + return url.toString() +} + +function buildGlobalEventWsUrl(lastEventId?: string): string { + const baseUrl = opencodeClient.getBaseUrl() + const normalizedBase = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/` + const httpUrl = new URL("global/event/ws", resolveAbsoluteUrl(normalizedBase)) + if (lastEventId && lastEventId.length > 0) { + httpUrl.searchParams.set("lastEventId", lastEventId) + } + return toWebSocketUrl(httpUrl.toString()) +} + type DirectoryQueue = { queue: Event[] buffer: Event[] @@ -92,11 +142,12 @@ type DirectoryQueue = { } export function createEventPipeline(input: EventPipelineInput) { - const { sdk, onEvent, onReconnect, routeDirectory } = input + const { sdk, onEvent, onReconnect, routeDirectory, transport = "auto" } = input const abort = new AbortController() let hasConnected = false + let lastEventId: string | undefined + let wsFallbackUntil = 0 - // One queue + one flush timer per directory. Lazily created on first event. const directories = new Map() const getOrCreateDir = (directory: string): DirectoryQueue => { @@ -114,19 +165,13 @@ export function createEventPipeline(input: EventPipelineInput) { return d } - // Coalesce key — same-type events for the same entity replace earlier ones. - // Keys are scoped to a single directory's queue, so directory is implicit. - // message.part.delta is a special case: consecutive deltas for the same - // (messageID, partID, field) are accumulated (string-concatenated) rather - // than replaced, because the reducer is a pure append and merging is - // semantically identical to applying each delta individually. const key = (payload: Event): string | undefined => { if (payload.type === "session.status") { const props = payload.properties as { sessionID: string } return `session.status:${props.sessionID}` } if (payload.type === "lsp.updated") { - return `lsp.updated` + return "lsp.updated" } if (payload.type === "message.part.updated") { const part = (payload.properties as { part: { messageID: string; id: string } }).part @@ -141,9 +186,6 @@ export function createEventPipeline(input: EventPipelineInput) { const deltaKey = (messageID: string, partID: string, field: string) => `${messageID}:${partID}:${field}` - // Flush one directory — swap queue, dispatch events. - // React 18 auto-batching still collapses the setState calls inside a single - // directory's flush into one render pass. const flushDir = (directory: string) => { const d = directories.get(directory) if (!d) return @@ -189,7 +231,6 @@ export function createEventPipeline(input: EventPipelineInput) { d.timer = setTimeout(() => flushDir(directory), Math.max(0, FLUSH_FRAME_MS - elapsed)) } - // Helpers const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) const isAbortError = (error: unknown): boolean => error instanceof DOMException && error.name === "AbortError" || @@ -200,6 +241,50 @@ export function createEventPipeline(input: EventPipelineInput) { let lastEventAt = Date.now() let heartbeat: ReturnType | undefined + const markConnected = () => { + if (hasConnected) { + onReconnect?.() + return + } + hasConnected = true + } + + const enqueueEvent = (directory: string, payload: Event) => { + const normalizedPayload = normalizeEventType(payload) + const routedDirectory = routeDirectory?.(directory, normalizedPayload) || directory + const d = getOrCreateDir(routedDirectory) + const k = key(normalizedPayload) + if (k) { + const i = d.coalesced.get(k) + if (i !== undefined) { + if (normalizedPayload.type === "message.part.delta") { + const prev = d.queue[i] as unknown as { properties: { delta: string } } + const inc = normalizedPayload.properties as { delta: string } + d.queue[i] = { + ...normalizedPayload, + properties: { + ...(normalizedPayload.properties as object), + delta: prev.properties.delta + inc.delta, + }, + } as unknown as Event + } else { + d.queue[i] = normalizedPayload + if (normalizedPayload.type === "message.part.updated") { + const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part + d.staleDeltas.add(deltaKey(part.messageID, part.id, "text")) + d.staleDeltas.add(deltaKey(part.messageID, part.id, "output")) + } + } + syncDebug.pipeline.coalesced(normalizedPayload.type, k) + return + } + d.coalesced.set(k, d.queue.length) + } + + d.queue.push(normalizedPayload) + scheduleDir(routedDirectory) + } + const resetHeartbeat = () => { lastEventAt = Date.now() if (heartbeat) clearTimeout(heartbeat) @@ -214,87 +299,214 @@ export function createEventPipeline(input: EventPipelineInput) { heartbeat = undefined } - // SSE loop — iterate SDK global event stream, enqueue with coalescing + const runSseAttempt = async (signal: AbortSignal) => { + const events = await sdk.global.event({ + signal, + onSseError: (error: unknown) => { + if (isAbortError(error)) return + if (streamErrorLogged) return + streamErrorLogged = true + console.error("[event-pipeline] SSE stream error", error) + }, + }) + + markConnected() + + let yielded = Date.now() + resetHeartbeat() + + for await (const event of events.stream) { + resetHeartbeat() + streamErrorLogged = false + const payload = resolveEventPayload((event as { payload?: Event }).payload ?? event) + if (!payload) { + continue + } + const directory = resolveEventDirectory(event, payload) + enqueueEvent(directory, payload) + + if (Date.now() - yielded < STREAM_YIELD_MS) continue + yielded = Date.now() + await wait(0) + } + } + + const runWsAttempt = async (signal: AbortSignal) => { + await new Promise((resolve, reject) => { + let settled = false + let opened = false + const socket = new WebSocket(buildGlobalEventWsUrl(lastEventId)) + const setFallbackCode = (error: Error) => { + if (!opened && transport === "auto") { + wsFallbackUntil = Date.now() + WS_FALLBACK_WINDOW_MS + ;(error as Error & { code?: string }).code = "WS_FALLBACK" + } + } + + let readyTimer: ReturnType | undefined = setTimeout(() => { + readyTimer = undefined + const error = new Error("Message stream WebSocket ready timeout") + setFallbackCode(error) + settleReject(error) + try { + socket.close() + } catch { + // ignore + } + }, WS_READY_TIMEOUT_MS) + + const cleanup = () => { + if (readyTimer) { + clearTimeout(readyTimer) + readyTimer = undefined + } + socket.onopen = null + socket.onmessage = null + socket.onerror = null + socket.onclose = null + } + + const settleResolve = () => { + if (settled) return + settled = true + signal.removeEventListener("abort", handleAbort) + cleanup() + resolve() + } + + const settleReject = (error: unknown) => { + if (settled) return + settled = true + signal.removeEventListener("abort", handleAbort) + cleanup() + reject(error) + } + + const handleAbort = () => { + try { + socket.close() + } catch { + // ignore close failures during abort + } + settleResolve() + } + + signal.addEventListener("abort", handleAbort, { once: true }) + + socket.onopen = () => { + streamErrorLogged = false + } + + socket.onmessage = (messageEvent) => { + resetHeartbeat() + streamErrorLogged = false + + let frame: MessageStreamWsFrame | null = null + try { + frame = JSON.parse(String(messageEvent.data)) as MessageStreamWsFrame + } catch (error) { + console.warn("[event-pipeline] Failed to parse WS frame", error) + return + } + + if (!frame || typeof frame.type !== "string") { + return + } + + if (frame.type === "ready") { + opened = true + if (readyTimer) { + clearTimeout(readyTimer) + readyTimer = undefined + } + markConnected() + return + } + + if (frame.type === "error") { + const error = new Error(frame.message || "Message stream WebSocket error") + setFallbackCode(error) + settleReject(error) + try { + socket.close() + } catch { + // ignore + } + return + } + + if (frame.type !== "event") { + return + } + + const payload = resolveEventPayload(frame.payload) + if (!payload) { + return + } + + if (typeof frame.eventId === "string" && frame.eventId.length > 0) { + lastEventId = frame.eventId + } + + const directory = resolveEventDirectory( + { directory: frame.directory, payload }, + payload, + ) + enqueueEvent(directory, payload) + } + + socket.onerror = () => { + void 0 + } + + socket.onclose = () => { + if (signal.aborted) { + settleResolve() + return + } + + const error = new Error("Global message stream WebSocket closed") + setFallbackCode(error) + settleReject(error) + } + }) + } + + const resolveTransport = (): "ws" | "sse" => { + if (typeof WebSocket !== "function") { + return "sse" + } + if (transport === "ws") { + return "ws" + } + if (transport === "sse") { + return "sse" + } + return wsFallbackUntil > Date.now() ? "sse" : "ws" + } + void (async () => { while (!abort.signal.aborted) { attempt = new AbortController() lastEventAt = Date.now() + let retryDelayMs = RECONNECT_DELAY_MS + const currentTransport = resolveTransport() const onAbort = () => { attempt?.abort() } abort.signal.addEventListener("abort", onAbort) try { - const events = await sdk.global.event({ - signal: attempt.signal, - onSseError: (error: unknown) => { - if (isAbortError(error)) return - if (streamErrorLogged) return - streamErrorLogged = true - console.error("[event-pipeline] stream error", error) - }, - }) - - if (hasConnected) { - onReconnect?.() + if (currentTransport === "ws") { + await runWsAttempt(attempt.signal) } else { - hasConnected = true - } - - let yielded = Date.now() - resetHeartbeat() - - // Enqueue event with coalescing + stale delta tracking - for await (const event of events.stream) { - resetHeartbeat() - streamErrorLogged = false - const payload = (event as { payload?: Event }).payload ?? (event as unknown as Event) - if (!payload || typeof payload !== "object" || typeof (payload as { type?: unknown }).type !== "string") { - continue - } - const normalizedPayload = normalizeEventType(payload) - const directory = resolveEventDirectory(event, normalizedPayload) - const routedDirectory = routeDirectory?.(directory, normalizedPayload) || directory - const d = getOrCreateDir(routedDirectory) - const k = key(normalizedPayload) - if (k) { - const i = d.coalesced.get(k) - if (i !== undefined) { - if (normalizedPayload.type === "message.part.delta") { - // Accumulate delta strings — append to the already-queued event - // rather than replacing it. The reducer is a pure string append so - // this is semantically identical to applying each delta separately. - const prev = d.queue[i] as unknown as { properties: { delta: string } } - const inc = normalizedPayload.properties as { delta: string } - d.queue[i] = { - ...normalizedPayload, - properties: { - ...(normalizedPayload.properties as object), - delta: prev.properties.delta + inc.delta, - }, - } as unknown as Event - } else { - d.queue[i] = normalizedPayload - if (normalizedPayload.type === "message.part.updated") { - const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part - d.staleDeltas.add(deltaKey(part.messageID, part.id, "text")) - d.staleDeltas.add(deltaKey(part.messageID, part.id, "output")) - } - } - syncDebug.pipeline.coalesced(normalizedPayload.type, k) - continue - } - d.coalesced.set(k, d.queue.length) - } - d.queue.push(normalizedPayload) - scheduleDir(routedDirectory) - - if (Date.now() - yielded < STREAM_YIELD_MS) continue - yielded = Date.now() - await wait(0) + await runSseAttempt(attempt.signal) } } catch (error) { - if (!isAbortError(error) && !streamErrorLogged) { + const code = typeof error === "object" && error !== null ? (error as { code?: unknown }).code : undefined + if (currentTransport === "ws" && code === "WS_FALLBACK") { + retryDelayMs = 0 + } else if (!isAbortError(error) && !streamErrorLogged) { streamErrorLogged = true console.error("[event-pipeline] stream failed", error) } @@ -305,12 +517,12 @@ export function createEventPipeline(input: EventPipelineInput) { } if (abort.signal.aborted) return - await wait(RECONNECT_DELAY_MS) + if (retryDelayMs > 0) { + await wait(retryDelayMs) + } } })().finally(flushAll) - // Visibility handler — abort SSE on heartbeat timeout so the loop reconnects. - // The reconnect triggers onReconnect above, which lets consumers resync state. const onVisibility = () => { if (typeof document === "undefined") return if (document.visibilityState !== "visible") return @@ -318,8 +530,6 @@ export function createEventPipeline(input: EventPipelineInput) { attempt?.abort() } - // pageshow handler — fires on back-forward cache restore (common on mobile PWA). - // bfcache restores the page without a fresh load, so SSE state may be stale. const onPageShow = (event: PageTransitionEvent) => { if (!event.persisted) return attempt?.abort() @@ -330,7 +540,6 @@ export function createEventPipeline(input: EventPipelineInput) { window.addEventListener("pageshow", onPageShow) } - // Cleanup — abort SSE, flush remaining events, remove listeners const cleanup = () => { if (typeof document !== "undefined") { document.removeEventListener("visibilitychange", onVisibility) diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index a7908806..f5fb6aed 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -26,6 +26,7 @@ import { stripMessageDiffSnapshots, stripSessionDiffSnapshots } from "./sanitize import { syncDebug } from "./debug" import { opencodeClient } from "@/lib/opencode/client" import { usePermissionStore } from "@/stores/permissionStore" +import { useConfigStore } from "@/stores/useConfigStore" import { toast } from "@/components/ui" import { appendNotification } from "./notification-store" import type { State } from "./types" @@ -1184,6 +1185,7 @@ export function SyncProvider(props: { directory: string children: React.ReactNode }) { + const messageStreamTransport = useConfigStore((state) => state.settingsMessageStreamTransport) const childStoresRef = useRef(null) if (!childStoresRef.current) childStoresRef.current = new ChildStoreManager() const childStores = childStoresRef.current @@ -1291,6 +1293,7 @@ export function SyncProvider(props: { const { cleanup } = createEventPipeline({ sdk: props.sdk, + transport: messageStreamTransport, routeDirectory: (directory, payload) => { return resolveDirectoryFromRoutingIndex(routingIndex, directory, payload, childStores) }, @@ -1314,7 +1317,7 @@ export function SyncProvider(props: { }, }) return cleanup - }, [props.sdk, props.directory, childStores, routingIndex]) + }, [props.sdk, childStores, routingIndex, messageStreamTransport]) // Ensure current directory's child store exists useEffect(() => { diff --git a/packages/web/server/index.js b/packages/web/server/index.js index b3836729..37284047 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -30,6 +30,10 @@ import { prepareNotificationLastMessage } from './lib/notifications/index.js'; import { registerTtsRoutes } from './lib/tts/routes.js'; import { detectSayTtsCapability } from './lib/tts/capability-runtime.js'; import { createTerminalRuntime } from './lib/terminal/runtime.js'; +import { + createGlobalUiEventBroadcaster, + createMessageStreamWsRuntime, +} from './lib/event-stream/index.js'; import { createFsSearchRuntime as createFsSearchRuntimeFactory } from './lib/fs/search.js'; import { createOpenCodeLifecycleRuntime } from './lib/opencode/lifecycle.js'; import { createOpenCodeEnvRuntime } from './lib/opencode/env-runtime.js'; @@ -76,6 +80,7 @@ const __dirname = path.dirname(__filename); const DEFAULT_PORT = 3000; const DESKTOP_NOTIFY_PREFIX = '[OpenChamberDesktopNotify] '; const uiNotificationClients = new Set(); +const uiNotificationWsClients = new Set(); const uiOpenChamberEventClients = new Set(); const HEALTH_CHECK_INTERVAL = 15000; const SHUTDOWN_TIMEOUT = 10000; @@ -306,15 +311,22 @@ const notificationEmitterRuntime = createNotificationEmitterRuntime({ getDesktopNotifyEnabled: () => ENV_DESKTOP_NOTIFY, desktopNotifyPrefix: DESKTOP_NOTIFY_PREFIX, getUiNotificationClients: () => uiNotificationClients, + getBroadcastGlobalUiEvent: () => broadcastGlobalUiEvent, }); const writeSseEvent = (...args) => notificationEmitterRuntime.writeSseEvent(...args); const emitDesktopNotification = (...args) => notificationEmitterRuntime.emitDesktopNotification(...args); +const broadcastGlobalUiEvent = createGlobalUiEventBroadcaster({ + sseClients: uiNotificationClients, + wsClients: uiNotificationWsClients, + writeSseEvent, +}); const broadcastUiNotification = (...args) => notificationEmitterRuntime.broadcastUiNotification(...args); const sessionRuntime = createSessionRuntime({ writeSseEvent, getNotificationClients: () => uiNotificationClients, + broadcastEvent: broadcastGlobalUiEvent, }); const projectConfigRuntime = createProjectConfigRuntime({ @@ -359,6 +371,7 @@ const tunnelAuthController = createTunnelAuth(); let runtimeManagedRemoteTunnelToken = ''; let runtimeManagedRemoteTunnelHostname = ''; let terminalRuntime = null; +let messageStreamRuntime = null; const userProvidedOpenCodePassword = hmrStateRuntime.getUserProvidedOpenCodePassword(hmrState); const initialOpenCodeAuthState = hmrStateRuntime.resolveOpenCodeAuthFromState({ hmrState, @@ -597,6 +610,50 @@ const openCodeWatcherRuntime = createOpenCodeWatcherRuntime({ }, }); +const processForwardedEventPayload = (payload, emitSyntheticEvent) => { + if (!payload || typeof payload !== 'object' || typeof emitSyntheticEvent !== 'function') { + return; + } + + maybeCacheSessionInfoFromEvent(payload); + + if (payload.type !== 'session.status') { + return; + } + + const properties = payload.properties && typeof payload.properties === 'object' ? payload.properties : {}; + const info = properties.info && typeof properties.info === 'object' ? properties.info : {}; + const sessionId = typeof properties.sessionID === 'string' ? properties.sessionID.trim() : ''; + const status = typeof info.type === 'string' ? info.type.trim() : ''; + + if (!sessionId || !status) { + return; + } + + emitSyntheticEvent({ + type: 'openchamber:session-status', + properties: { + sessionId, + status, + timestamp: Date.now(), + metadata: { + attempt: typeof info.attempt === 'number' ? info.attempt : undefined, + message: typeof info.message === 'string' ? info.message : undefined, + next: typeof info.next === 'number' ? info.next : undefined, + }, + needsAttention: false, + }, + }); + + emitSyntheticEvent({ + type: 'openchamber:session-activity', + properties: { + sessionId, + phase: status === 'busy' || status === 'retry' ? 'busy' : 'idle', + }, + }); +}; + const serverUtilsRuntime = createServerUtilsRuntime({ fs, @@ -704,6 +761,7 @@ const tunnelWiringRuntime = createTunnelWiringRuntime({ }); const startupPipelineRuntime = createStartupPipelineRuntime({ createTerminalRuntime, + createMessageStreamWsRuntime, createServerStartupRuntime, }); @@ -842,6 +900,10 @@ const gracefulShutdownRuntime = createGracefulShutdownRuntime({ setTerminalRuntime: (value) => { terminalRuntime = value; }, + getMessageStreamRuntime: () => messageStreamRuntime, + setMessageStreamRuntime: (value) => { + messageStreamRuntime = value; + }, shouldSkipOpenCodeStop: () => ENV_SKIP_OPENCODE_START || isExternalOpenCode, getOpenCodePort: () => openCodePort, getOpenCodeProcess: () => openCodeProcess, @@ -1028,6 +1090,10 @@ async function main(options = {}) { isExecutable, isRequestOriginAllowed, rejectWebSocketUpgrade, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + processForwardedEventPayload, + messageStreamWsClients: uiNotificationWsClients, terminalHeartbeatIntervalMs: TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS, terminalRebindWindowMs: TERMINAL_INPUT_WS_REBIND_WINDOW_MS, terminalMaxRebindsPerWindow: TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW, @@ -1058,6 +1124,7 @@ async function main(options = {}) { attachSignals, }); terminalRuntime = startupPipelineResult.terminalRuntime; + messageStreamRuntime = startupPipelineResult.messageStreamRuntime; try { await scheduledTasksRuntime.start(); diff --git a/packages/web/server/lib/event-stream/DOCUMENTATION.md b/packages/web/server/lib/event-stream/DOCUMENTATION.md new file mode 100644 index 00000000..f237c83b --- /dev/null +++ b/packages/web/server/lib/event-stream/DOCUMENTATION.md @@ -0,0 +1,43 @@ +# Event Stream Module Documentation + +## Purpose +This module contains the OpenChamber message-stream WebSocket protocol and runtime bridge. It keeps the browser-facing WebSocket transport separate from the upstream OpenCode SSE transport. + +## 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/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/protocol.test.js`: unit tests for protocol helpers. +- `packages/web/server/lib/event-stream/runtime.test.js`: unit tests for runtime-side broadcaster behavior. + +## Public exports + +### Protocol helpers +- `MESSAGE_STREAM_GLOBAL_WS_PATH`: `/api/global/event/ws` +- `MESSAGE_STREAM_DIRECTORY_WS_PATH`: `/api/event/ws` +- `MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS`: heartbeat interval for browser-facing WS connections. +- `parseSseEventEnvelope(block)`: parses an SSE block into `{ eventId, directory, payload }`. +- `sendMessageStreamWsFrame(socket, payload)`: serializes and sends a JSON WS frame. +- `sendMessageStreamWsEvent(socket, payload, options)`: sends an event frame with optional `eventId` and `directory`. + +### Runtime helpers +- `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. + +## 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. +- Global synthetic events such as `openchamber:session-status`, `openchamber:session-activity`, `openchamber:notification`, and `openchamber:heartbeat` are preserved on the WS path. +- Global UI broadcasts are fan-out capable across both SSE and WS clients. + +## 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`. + +## Testing +- Run `bun test packages/web/server/lib/event-stream/protocol.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/index.js b/packages/web/server/lib/event-stream/index.js new file mode 100644 index 00000000..ec6e3fa3 --- /dev/null +++ b/packages/web/server/lib/event-stream/index.js @@ -0,0 +1,13 @@ +export { + MESSAGE_STREAM_GLOBAL_WS_PATH, + MESSAGE_STREAM_DIRECTORY_WS_PATH, + MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS, + parseSseEventEnvelope, + sendMessageStreamWsFrame, + sendMessageStreamWsEvent, +} from './protocol.js'; + +export { + createGlobalUiEventBroadcaster, + createMessageStreamWsRuntime, +} from './runtime.js'; diff --git a/packages/web/server/lib/event-stream/protocol.js b/packages/web/server/lib/event-stream/protocol.js new file mode 100644 index 00000000..033859db --- /dev/null +++ b/packages/web/server/lib/event-stream/protocol.js @@ -0,0 +1,82 @@ +export const MESSAGE_STREAM_GLOBAL_WS_PATH = '/api/global/event/ws'; +export const MESSAGE_STREAM_DIRECTORY_WS_PATH = '/api/event/ws'; +export const MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS = 15 * 1000; + +export function parseSseEventEnvelope(block) { + if (!block || typeof block !== 'string') { + return null; + } + + const eventId = block + .split('\n') + .find((line) => line.startsWith('id:')) + ?.slice(3) + .trim() || 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); + if ( + parsed && + typeof parsed === 'object' && + typeof parsed.payload === 'object' && + parsed.payload !== null + ) { + return { + eventId, + directory: typeof parsed.directory === 'string' && parsed.directory.length > 0 ? parsed.directory : null, + payload: parsed.payload, + }; + } + + const directory = + typeof parsed?.directory === 'string' && parsed.directory.length > 0 + ? parsed.directory + : typeof parsed?.properties?.directory === 'string' && parsed.properties.directory.length > 0 + ? parsed.properties.directory + : null; + + return { + eventId, + directory, + payload: parsed, + }; + } catch { + return null; + } +} + +export function sendMessageStreamWsFrame(socket, payload) { + if (!socket || socket.readyState !== 1) { + return false; + } + + try { + socket.send(JSON.stringify(payload)); + return true; + } catch { + return false; + } +} + +export function sendMessageStreamWsEvent(socket, payload, options = {}) { + return sendMessageStreamWsFrame(socket, { + type: 'event', + payload, + ...(typeof options.eventId === 'string' && options.eventId.length > 0 ? { eventId: options.eventId } : {}), + ...(typeof options.directory === 'string' && options.directory.length > 0 ? { directory: options.directory } : {}), + }); +} diff --git a/packages/web/server/lib/event-stream/protocol.test.js b/packages/web/server/lib/event-stream/protocol.test.js new file mode 100644 index 00000000..62eec900 --- /dev/null +++ b/packages/web/server/lib/event-stream/protocol.test.js @@ -0,0 +1,89 @@ +import { describe, expect, it } from 'bun:test'; + +import { + MESSAGE_STREAM_DIRECTORY_WS_PATH, + MESSAGE_STREAM_GLOBAL_WS_PATH, + parseSseEventEnvelope, + sendMessageStreamWsEvent, + sendMessageStreamWsFrame, +} from './protocol.js'; + +describe('event stream protocol helpers', () => { + it('exports stable websocket paths', () => { + expect(MESSAGE_STREAM_GLOBAL_WS_PATH).toBe('/api/global/event/ws'); + expect(MESSAGE_STREAM_DIRECTORY_WS_PATH).toBe('/api/event/ws'); + }); + + it('parses wrapped SSE payloads with event id and directory', () => { + const envelope = parseSseEventEnvelope( + 'id: evt-1\n' + + 'event: message\n' + + 'data: {"directory":"/tmp/project","payload":{"type":"session.updated"}}\n' + ); + + expect(envelope).toEqual({ + eventId: 'evt-1', + directory: '/tmp/project', + payload: { type: 'session.updated' }, + }); + }); + + it('derives directory from payload properties when not wrapped', () => { + const envelope = parseSseEventEnvelope( + 'data: {"type":"openchamber:notification","properties":{"directory":"/tmp/project"}}\n' + ); + + expect(envelope).toEqual({ + eventId: null, + directory: '/tmp/project', + payload: { + type: 'openchamber:notification', + properties: { directory: '/tmp/project' }, + }, + }); + }); + + it('returns null for malformed SSE blocks', () => { + expect(parseSseEventEnvelope('event: message\n')).toBeNull(); + expect(parseSseEventEnvelope('data: {oops}\n')).toBeNull(); + }); + + it('serializes generic websocket frames', () => { + let rawPayload = null; + const socket = { + readyState: 1, + send(payload) { + rawPayload = payload; + }, + }; + + const sent = sendMessageStreamWsFrame(socket, { type: 'ready' }); + + expect(sent).toBe(true); + expect(rawPayload).toBe('{"type":"ready"}'); + }); + + it('serializes event frames with routing metadata', () => { + let rawPayload = null; + const socket = { + readyState: 1, + send(payload) { + rawPayload = payload; + }, + }; + + const sent = sendMessageStreamWsEvent( + socket, + { type: 'openchamber:heartbeat', timestamp: 1 }, + { eventId: 'evt-2', directory: '/tmp/project' } + ); + + expect(sent).toBe(true); + expect(JSON.parse(rawPayload)).toEqual({ + type: 'event', + payload: { type: 'openchamber:heartbeat', timestamp: 1 }, + eventId: 'evt-2', + directory: '/tmp/project', + }); + }); +}); diff --git a/packages/web/server/lib/event-stream/runtime.js b/packages/web/server/lib/event-stream/runtime.js new file mode 100644 index 00000000..a25895dc --- /dev/null +++ b/packages/web/server/lib/event-stream/runtime.js @@ -0,0 +1,288 @@ +import { WebSocketServer } from 'ws'; + +import { parseRequestPathname } from '../terminal/index.js'; +import { + MESSAGE_STREAM_DIRECTORY_WS_PATH, + MESSAGE_STREAM_GLOBAL_WS_PATH, + MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS, + parseSseEventEnvelope, + sendMessageStreamWsEvent, + sendMessageStreamWsFrame, +} from './protocol.js'; + +export function createGlobalUiEventBroadcaster({ + sseClients, + wsClients, + writeSseEvent, +}) { + return (payload, options = {}) => { + const hasSseClients = sseClients.size > 0; + const hasWsClients = wsClients.size > 0; + if (!hasSseClients && !hasWsClients) { + return; + } + + if (hasSseClients) { + for (const res of sseClients) { + try { + writeSseEvent(res, payload); + } catch { + } + } + } + + if (hasWsClients) { + for (const socket of Array.from(wsClients)) { + const sent = sendMessageStreamWsEvent(socket, payload, { + directory: typeof options.directory === 'string' && options.directory.length > 0 ? options.directory : 'global', + eventId: typeof options.eventId === 'string' && options.eventId.length > 0 ? options.eventId : undefined, + }); + if (!sent) { + wsClients.delete(socket); + } + } + } + }; +} + +export function createMessageStreamWsRuntime({ + server, + uiAuthController, + isRequestOriginAllowed, + rejectWebSocketUpgrade, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + processForwardedEventPayload, + wsClients, + fetchImpl = fetch, +}) { + const wsServer = new WebSocketServer({ + noServer: true, + }); + + wsServer.on('connection', (socket, req) => { + const rawUrl = typeof req?.url === 'string' ? req.url : MESSAGE_STREAM_GLOBAL_WS_PATH; + const pathname = parseRequestPathname(rawUrl); + const requestUrl = new URL(rawUrl, 'http://127.0.0.1'); + const isGlobalStream = pathname === MESSAGE_STREAM_GLOBAL_WS_PATH; + const requestedLastEventId = requestUrl.searchParams.get('lastEventId')?.trim() || ''; + const requestedDirectory = requestUrl.searchParams.get('directory')?.trim() || ''; + + const controller = new AbortController(); + const cleanup = () => { + if (!controller.signal.aborted) { + controller.abort(); + } + wsClients.delete(socket); + }; + + const pingInterval = setInterval(() => { + if (socket.readyState !== 1) { + return; + } + + try { + socket.ping(); + } catch { + } + }, MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS); + + const heartbeatInterval = setInterval(() => { + sendMessageStreamWsEvent(socket, { type: 'openchamber:heartbeat', timestamp: Date.now() }, { directory: 'global' }); + }, MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS); + + socket.on('close', () => { + clearInterval(pingInterval); + clearInterval(heartbeatInterval); + cleanup(); + }); + + socket.on('error', () => { + void 0; + }); + + const run = async () => { + 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; + } + + if (!isGlobalStream && requestedDirectory) { + targetUrl.searchParams.set('directory', requestedDirectory); + } + + const headers = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + ...getOpenCodeAuthHeaders(), + }; + + if (requestedLastEventId) { + headers['Last-Event-ID'] = requestedLastEventId; + } + + let upstream; + try { + upstream = await fetchImpl(targetUrl.toString(), { + headers, + signal: controller.signal, + }); + } catch { + if (!controller.signal.aborted) { + sendMessageStreamWsFrame(socket, { type: 'error', message: 'Failed to connect to OpenCode event stream' }); + socket.close(1011, 'Failed to connect to OpenCode event stream'); + } + return; + } + + if (!upstream.ok || !upstream.body) { + sendMessageStreamWsFrame(socket, { + type: 'error', + message: `OpenCode event stream unavailable (${upstream.status})`, + }); + socket.close(1011, 'OpenCode event stream unavailable'); + return; + } + + sendMessageStreamWsFrame(socket, { + type: 'ready', + scope: isGlobalStream ? 'global' : 'directory', + }); + + if (isGlobalStream) { + wsClients.add(socket); + } + + const decoder = new TextDecoder(); + const reader = upstream.body.getReader(); + let buffer = ''; + + const forwardBlock = (block) => { + if (!block) { + return; + } + + const envelope = parseSseEventEnvelope(block); + const payload = envelope?.payload ?? null; + if (!payload) { + return; + } + + const directory = isGlobalStream + ? (typeof envelope?.directory === 'string' && envelope.directory.length > 0 ? envelope.directory : 'global') + : (requestedDirectory || envelope?.directory || 'global'); + + sendMessageStreamWsEvent(socket, payload, { + directory, + eventId: typeof envelope?.eventId === 'string' && envelope.eventId.length > 0 ? envelope.eventId : undefined, + }); + + processForwardedEventPayload(payload, (syntheticPayload) => { + sendMessageStreamWsEvent(socket, syntheticPayload, { directory: 'global' }); + }); + }; + + 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 = 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) { + console.warn('Message stream WS proxy error:', error); + sendMessageStreamWsFrame(socket, { type: 'error', message: 'Message stream proxy error' }); + socket.close(1011, 'Message stream proxy error'); + } + } finally { + cleanup(); + try { + if (socket.readyState === 1 || socket.readyState === 0) { + socket.close(); + } + } catch { + } + } + }; + + void run(); + }); + + const upgradeHandler = (req, socket, head) => { + const pathname = parseRequestPathname(req.url); + if (pathname !== MESSAGE_STREAM_GLOBAL_WS_PATH && pathname !== MESSAGE_STREAM_DIRECTORY_WS_PATH) { + return; + } + + const handleUpgrade = async () => { + try { + if (uiAuthController?.enabled) { + const sessionToken = await uiAuthController?.ensureSessionToken?.(req, null); + if (!sessionToken) { + rejectWebSocketUpgrade(socket, 401, 'UI authentication required'); + return; + } + + const originAllowed = await isRequestOriginAllowed(req); + if (!originAllowed) { + rejectWebSocketUpgrade(socket, 403, 'Invalid origin'); + return; + } + } + + wsServer.handleUpgrade(req, socket, head, (ws) => { + wsServer.emit('connection', ws, req); + }); + } catch { + rejectWebSocketUpgrade(socket, 500, 'Upgrade failed'); + } + }; + + void handleUpgrade(); + }; + + server.on('upgrade', upgradeHandler); + + return { + wsServer, + async close() { + server.off('upgrade', upgradeHandler); + + try { + for (const client of wsServer.clients) { + try { + client.terminate(); + } catch { + } + } + + await new Promise((resolve) => { + wsServer.close(() => resolve()); + }); + } catch { + } finally { + wsClients.clear(); + } + }, + }; +} diff --git a/packages/web/server/lib/event-stream/runtime.test.js b/packages/web/server/lib/event-stream/runtime.test.js new file mode 100644 index 00000000..7b6b416e --- /dev/null +++ b/packages/web/server/lib/event-stream/runtime.test.js @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'bun:test'; + +import { createGlobalUiEventBroadcaster } from './runtime.js'; + +describe('event stream broadcaster', () => { + it('fans out synthetic events to SSE and WS clients', () => { + const sseEvents = []; + const wsPayloads = []; + const sseClient = { id: 'sse-1' }; + const wsClient = { + readyState: 1, + send(payload) { + wsPayloads.push(JSON.parse(payload)); + }, + }; + + const broadcast = createGlobalUiEventBroadcaster({ + sseClients: new Set([sseClient]), + wsClients: new Set([wsClient]), + writeSseEvent(res, payload) { + sseEvents.push({ res, payload }); + }, + }); + + broadcast({ type: 'openchamber:session-status' }, { eventId: 'evt-1', directory: '/tmp/project' }); + + expect(sseEvents).toEqual([ + { + res: sseClient, + payload: { type: 'openchamber:session-status' }, + }, + ]); + expect(wsPayloads).toEqual([ + { + type: 'event', + payload: { type: 'openchamber:session-status' }, + eventId: 'evt-1', + directory: '/tmp/project', + }, + ]); + }); + + it('removes websocket clients that fail to receive a payload', () => { + const wsClients = new Set([ + { + readyState: 1, + send() { + throw new Error('socket write failed'); + }, + }, + ]); + + const broadcast = createGlobalUiEventBroadcaster({ + sseClients: new Set(), + wsClients, + writeSseEvent() { + throw new Error('should not be called'); + }, + }); + + broadcast({ type: 'openchamber:notification' }); + + expect(wsClients.size).toBe(0); + }); +}); diff --git a/packages/web/server/lib/notifications/emitter-runtime.js b/packages/web/server/lib/notifications/emitter-runtime.js index 2ffc9cfe..8bded97b 100644 --- a/packages/web/server/lib/notifications/emitter-runtime.js +++ b/packages/web/server/lib/notifications/emitter-runtime.js @@ -4,6 +4,7 @@ export const createNotificationEmitterRuntime = (dependencies) => { getDesktopNotifyEnabled, desktopNotifyPrefix, getUiNotificationClients, + getBroadcastGlobalUiEvent, } = dependencies; const writeSseEvent = (res, payload) => { @@ -34,6 +35,25 @@ export const createNotificationEmitterRuntime = (dependencies) => { return; } + const syntheticPayload = { + type: 'openchamber:notification', + properties: { + ...payload, + // Tell the UI whether the sidecar stdout notification channel is active. + // When true, the desktop UI should skip this SSE notification to avoid duplicates. + // When false (e.g. tauri dev), the UI must handle this SSE notification itself. + desktopStdoutActive: desktopNotifyEnabled, + }, + }; + + const broadcastGlobalUiEvent = typeof getBroadcastGlobalUiEvent === 'function' + ? getBroadcastGlobalUiEvent() + : null; + if (broadcastGlobalUiEvent) { + broadcastGlobalUiEvent(syntheticPayload); + return; + } + const clients = getUiNotificationClients(); if (clients.size === 0) { return; @@ -41,16 +61,7 @@ export const createNotificationEmitterRuntime = (dependencies) => { for (const res of clients) { try { - writeSseEvent(res, { - type: 'openchamber:notification', - properties: { - ...payload, - // Tell the UI whether the sidecar stdout notification channel is active. - // When true, the desktop UI should skip this SSE notification to avoid duplicates. - // When false (e.g. tauri dev), the UI must handle this SSE notification itself. - desktopStdoutActive: desktopNotifyEnabled, - }, - }); + writeSseEvent(res, syntheticPayload); } catch { // ignore } diff --git a/packages/web/server/lib/opencode/DOCUMENTATION.md b/packages/web/server/lib/opencode/DOCUMENTATION.md index e5cfb40d..f58fdce2 100644 --- a/packages/web/server/lib/opencode/DOCUMENTATION.md +++ b/packages/web/server/lib/opencode/DOCUMENTATION.md @@ -80,7 +80,7 @@ This module provides OpenCode server integration utilities for the web server ru - Keeps route behavior independent from composition root; `index.js` now supplies dependencies only. ## Public exports (session-runtime.js) -- `createSessionRuntime({ writeSseEvent, getNotificationClients })`: creates runtime-owned state machine and APIs for session status. +- `createSessionRuntime({ writeSseEvent, getNotificationClients, broadcastEvent? })`: creates runtime-owned state machine and APIs for session status. - Returned API: - `processOpenCodeSsePayload(payload)` - `getSessionActivitySnapshot()` diff --git a/packages/web/server/lib/opencode/session-runtime.js b/packages/web/server/lib/opencode/session-runtime.js index 0f181281..48a4c9ad 100644 --- a/packages/web/server/lib/opencode/session-runtime.js +++ b/packages/web/server/lib/opencode/session-runtime.js @@ -42,7 +42,7 @@ const deriveSessionActivityTransitions = (payload) => { return []; }; -export const createSessionRuntime = ({ writeSseEvent, getNotificationClients }) => { +export const createSessionRuntime = ({ writeSseEvent, getNotificationClients, broadcastEvent }) => { const sessionActivityPhases = new Map(); const sessionActivityCooldowns = new Map(); const sessionStates = new Map(); @@ -93,6 +93,16 @@ export const createSessionRuntime = ({ writeSseEvent, getNotificationClients }) sessionActivityCooldowns.set(sessionId, timer); } + if (typeof broadcastEvent === 'function') { + broadcastEvent({ + type: 'openchamber:session-activity', + properties: { + sessionId, + phase, + }, + }); + } + return true; }; @@ -132,21 +142,27 @@ export const createSessionRuntime = ({ writeSseEvent, getNotificationClients }) const attentionState = sessionAttentionStates.get(sessionId); const attentionChanged = !!attentionState && existingAttentionState?.needsAttention !== attentionState.needsAttention; const clients = getNotificationClients(); - if (clients.size > 0 && (!existing || existing.status !== status || attentionChanged)) { + if (!existing || existing.status !== status || attentionChanged) { const state = sessionStates.get(sessionId); - for (const res of clients) { - try { - writeSseEvent(res, { - type: 'openchamber:session-status', - properties: { - sessionId, - status: state.status, - timestamp: state.lastUpdateAt, - metadata: state.metadata, - needsAttention: attentionState?.needsAttention ?? false, - }, - }); - } catch { + const syntheticPayload = { + type: 'openchamber:session-status', + properties: { + sessionId, + status: state.status, + timestamp: state.lastUpdateAt, + metadata: state.metadata, + needsAttention: attentionState?.needsAttention ?? false, + }, + }; + + if (typeof broadcastEvent === 'function') { + broadcastEvent(syntheticPayload); + } else if (clients.size > 0) { + for (const res of clients) { + try { + writeSseEvent(res, syntheticPayload); + } catch { + } } } } @@ -183,20 +199,27 @@ export const createSessionRuntime = ({ writeSseEvent, getNotificationClients }) if (wasNeedsAttention) { state.needsAttention = false; - const clients = getNotificationClients(); - for (const res of clients) { - try { - writeSseEvent(res, { - type: 'openchamber:session-status', - properties: { - sessionId, - status: state.status, - timestamp: Date.now(), - metadata: {}, - needsAttention: false, - }, - }); - } catch { + + const syntheticPayload = { + type: 'openchamber:session-status', + properties: { + sessionId, + status: state.status, + timestamp: Date.now(), + metadata: {}, + needsAttention: false, + }, + }; + + if (typeof broadcastEvent === 'function') { + broadcastEvent(syntheticPayload); + } else { + const clients = getNotificationClients(); + for (const res of clients) { + try { + writeSseEvent(res, syntheticPayload); + } catch { + } } } } diff --git a/packages/web/server/lib/opencode/session-runtime.test.js b/packages/web/server/lib/opencode/session-runtime.test.js new file mode 100644 index 00000000..42224cae --- /dev/null +++ b/packages/web/server/lib/opencode/session-runtime.test.js @@ -0,0 +1,68 @@ +import { afterEach, describe, expect, it } from 'bun:test'; + +import { createSessionRuntime } from './session-runtime.js'; + +describe('session runtime', () => { + const runtimes = []; + + afterEach(() => { + for (const runtime of runtimes) { + runtime.dispose(); + } + runtimes.length = 0; + }); + + it('broadcasts attention clears through the shared broadcaster', () => { + const events = []; + const runtime = createSessionRuntime({ + writeSseEvent() { + throw new Error('SSE fallback should not be used when broadcastEvent is provided'); + }, + getNotificationClients: () => new Set(), + broadcastEvent: (payload) => { + events.push(payload); + }, + }); + runtimes.push(runtime); + + runtime.processOpenCodeSsePayload({ + type: 'session.status', + properties: { + sessionID: 'session-1', + info: { + type: 'busy', + }, + }, + }); + runtime.markUserMessageSent('session-1'); + runtime.processOpenCodeSsePayload({ + type: 'session.status', + properties: { + sessionID: 'session-1', + info: { + type: 'idle', + }, + }, + }); + runtime.markSessionViewed('session-1', 'client-1'); + + expect(events).toContainEqual({ + type: 'openchamber:session-status', + properties: expect.objectContaining({ + sessionId: 'session-1', + status: 'idle', + needsAttention: true, + }), + }); + expect(events.at(-1)).toEqual({ + type: 'openchamber:session-status', + properties: { + sessionId: 'session-1', + status: 'idle', + timestamp: expect.any(Number), + metadata: {}, + needsAttention: false, + }, + }); + }); +}); diff --git a/packages/web/server/lib/opencode/settings-helpers.js b/packages/web/server/lib/opencode/settings-helpers.js index 2655521b..d6ee84c3 100644 --- a/packages/web/server/lib/opencode/settings-helpers.js +++ b/packages/web/server/lib/opencode/settings-helpers.js @@ -325,6 +325,12 @@ export const createSettingsHelpers = (dependencies) => { result.chatRenderMode = mode; } } + if (typeof candidate.messageStreamTransport === 'string') { + const mode = candidate.messageStreamTransport.trim(); + if (mode === 'auto' || mode === 'ws' || mode === 'sse') { + result.messageStreamTransport = mode; + } + } if (typeof candidate.activityRenderMode === 'string') { const mode = candidate.activityRenderMode.trim(); if (mode === 'collapsed' || mode === 'summary') { diff --git a/packages/web/server/lib/opencode/settings-helpers.test.js b/packages/web/server/lib/opencode/settings-helpers.test.js new file mode 100644 index 00000000..d03e68df --- /dev/null +++ b/packages/web/server/lib/opencode/settings-helpers.test.js @@ -0,0 +1,43 @@ +import { describe, expect, it } from 'bun:test'; + +import { createSettingsHelpers } from './settings-helpers.js'; + +const createTestHelpers = () => createSettingsHelpers({ + normalizePathForPersistence: (value) => value, + normalizeDirectoryPath: (value) => value, + normalizeTunnelBootstrapTtlMs: (value) => value, + normalizeTunnelSessionTtlMs: (value) => value, + normalizeTunnelProvider: (value) => value, + normalizeTunnelMode: (value) => value, + normalizeOptionalPath: (value) => value, + normalizeManagedRemoteTunnelHostname: (value) => value, + normalizeManagedRemoteTunnelPresets: () => undefined, + normalizeManagedRemoteTunnelPresetTokens: () => undefined, + sanitizeTypographySizesPartial: () => undefined, + normalizeStringArray: (input) => input, + sanitizeModelRefs: () => undefined, + sanitizeSkillCatalogs: () => undefined, + sanitizeProjects: () => undefined, +}); + +describe('settings helpers', () => { + it('accepts messageStreamTransport as a persisted shared setting', () => { + const helpers = createTestHelpers(); + + expect(helpers.sanitizeSettingsUpdate({ messageStreamTransport: 'ws' })).toEqual({ + messageStreamTransport: 'ws', + }); + expect(helpers.sanitizeSettingsUpdate({ messageStreamTransport: 'sse' })).toEqual({ + messageStreamTransport: 'sse', + }); + expect(helpers.sanitizeSettingsUpdate({ messageStreamTransport: 'auto' })).toEqual({ + messageStreamTransport: 'auto', + }); + }); + + it('rejects invalid messageStreamTransport values', () => { + const helpers = createTestHelpers(); + + expect(helpers.sanitizeSettingsUpdate({ messageStreamTransport: 'websocket' })).toEqual({}); + }); +}); diff --git a/packages/web/server/lib/opencode/shutdown-runtime.js b/packages/web/server/lib/opencode/shutdown-runtime.js index 227bf5db..e339cb3e 100644 --- a/packages/web/server/lib/opencode/shutdown-runtime.js +++ b/packages/web/server/lib/opencode/shutdown-runtime.js @@ -13,6 +13,8 @@ export const createGracefulShutdownRuntime = (dependencies) => { clearHealthCheckInterval, getTerminalRuntime, setTerminalRuntime, + getMessageStreamRuntime, + setMessageStreamRuntime, shouldSkipOpenCodeStop, getOpenCodePort, getOpenCodeProcess, @@ -54,6 +56,16 @@ export const createGracefulShutdownRuntime = (dependencies) => { } } + const messageStreamRuntime = getMessageStreamRuntime(); + if (messageStreamRuntime) { + try { + await messageStreamRuntime.close(); + } catch { + } finally { + setMessageStreamRuntime(null); + } + } + if (!shouldSkipOpenCodeStop()) { const portToKill = getOpenCodePort(); const openCodeProcess = getOpenCodeProcess(); diff --git a/packages/web/server/lib/opencode/startup-pipeline-runtime.js b/packages/web/server/lib/opencode/startup-pipeline-runtime.js index f7227e13..697f0d3c 100644 --- a/packages/web/server/lib/opencode/startup-pipeline-runtime.js +++ b/packages/web/server/lib/opencode/startup-pipeline-runtime.js @@ -1,6 +1,7 @@ export const createStartupPipelineRuntime = (dependencies) => { const { createTerminalRuntime, + createMessageStreamWsRuntime, createServerStartupRuntime, } = dependencies; @@ -17,6 +18,10 @@ export const createStartupPipelineRuntime = (dependencies) => { isExecutable, isRequestOriginAllowed, rejectWebSocketUpgrade, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + processForwardedEventPayload, + messageStreamWsClients, terminalHeartbeatIntervalMs, terminalRebindWindowMs, terminalMaxRebindsPerWindow, @@ -62,6 +67,17 @@ export const createStartupPipelineRuntime = (dependencies) => { TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW: terminalMaxRebindsPerWindow, }); + const messageStreamRuntime = createMessageStreamWsRuntime({ + server, + uiAuthController, + isRequestOriginAllowed, + rejectWebSocketUpgrade, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + processForwardedEventPayload, + wsClients: messageStreamWsClients, + }); + setupProxy(app); scheduleOpenCodeApiDetection(); void bootstrapOpenCodeAtStartup(); @@ -98,6 +114,7 @@ export const createStartupPipelineRuntime = (dependencies) => { return { terminalRuntime, + messageStreamRuntime, }; };