diff --git a/packages/desktop/src-tauri/src/main.rs b/packages/desktop/src-tauri/src/main.rs index 831648d5..84630b07 100644 --- a/packages/desktop/src-tauri/src/main.rs +++ b/packages/desktop/src-tauri/src/main.rs @@ -2068,16 +2068,23 @@ async fn spawn_local_server(app: &tauri::AppHandle) -> Result { let app_handle = app.clone(); tauri::async_runtime::spawn(async move { let mut rx = rx; + let mut stdout_buffer = String::new(); while let Some(event) = rx.recv().await { match event { CommandEvent::Stdout(bytes) => { - let line = String::from_utf8_lossy(&bytes); - if let Some(rest) = line.strip_prefix(SIDECAR_NOTIFY_PREFIX) { - if let Ok(parsed) = - serde_json::from_str::(rest.trim()) - { - maybe_show_sidecar_notification(&app_handle, parsed); + stdout_buffer.push_str(&String::from_utf8_lossy(&bytes)); + + while let Some(newline_index) = stdout_buffer.find('\n') { + let line = stdout_buffer[..newline_index].trim_end_matches('\r'); + if let Some(rest) = line.strip_prefix(SIDECAR_NOTIFY_PREFIX) { + if let Ok(parsed) = + serde_json::from_str::(rest.trim()) + { + maybe_show_sidecar_notification(&app_handle, parsed); + } } + + stdout_buffer.drain(..=newline_index); } } CommandEvent::Error(error) => { diff --git a/packages/ui/src/App.tsx b/packages/ui/src/App.tsx index 11156d9c..57507122 100644 --- a/packages/ui/src/App.tsx +++ b/packages/ui/src/App.tsx @@ -175,6 +175,7 @@ function App({ apis }: AppProps) { const appReadyDispatchedRef = React.useRef(false); const embeddedSessionChat = React.useMemo(() => readEmbeddedSessionChatConfig(), []); const embeddedBackgroundWorkEnabled = !embeddedSessionChat || isEmbeddedVisible; + const recentDesktopNotificationTagsRef = React.useRef>(new Map()); React.useEffect(() => { setStreamPerfEnabled(showMemoryDebug); @@ -371,6 +372,63 @@ function App({ apis }: AppProps) { }; }, [embeddedSessionChat]); + React.useEffect(() => { + if (embeddedSessionChat || !isDesktopRuntime || typeof window === 'undefined' || typeof EventSource === 'undefined') { + return; + } + + const source = new EventSource('/api/notifications/stream'); + + const handleMessage = (event: MessageEvent) => { + type DesktopNotificationEvent = { + type?: string; + properties?: { + title?: string; + body?: string; + tag?: string; + }; + }; + + let payload: DesktopNotificationEvent; + + try { + payload = JSON.parse(event.data) as DesktopNotificationEvent; + } catch { + return; + } + + if (payload?.type !== 'openchamber:notification') { + return; + } + + const tag = typeof payload.properties?.tag === 'string' ? payload.properties.tag : ''; + if (tag) { + const now = Date.now(); + const lastSeenAt = recentDesktopNotificationTagsRef.current.get(tag) ?? 0; + if (now - lastSeenAt < 5000) { + return; + } + recentDesktopNotificationTagsRef.current.set(tag, now); + } + + void apis.notifications.notifyAgentCompletion({ + title: payload.properties?.title, + body: payload.properties?.body, + tag: tag || undefined, + }); + }; + + source.addEventListener('message', handleMessage as EventListener); + source.onerror = () => { + // Let EventSource reconnect automatically. + }; + + return () => { + source.removeEventListener('message', handleMessage as EventListener); + source.close(); + }; + }, [apis.notifications, embeddedSessionChat, isDesktopRuntime]); + React.useEffect(() => { if (!embeddedSessionChat?.directory || isVSCodeRuntime) { return; diff --git a/packages/ui/src/components/chat/ChatContainer.tsx b/packages/ui/src/components/chat/ChatContainer.tsx index 073a548f..d7c2d532 100644 --- a/packages/ui/src/components/chat/ChatContainer.tsx +++ b/packages/ui/src/components/chat/ChatContainer.tsx @@ -631,6 +631,7 @@ export const ChatContainer: React.FC = () => { }, [currentSessionId, isDesktopExpandedInput, scrollRef]); const hasHistoryMetadata = Boolean(historyMeta); + const lastHydratedSessionRef = React.useRef(null); const isSessionHydrating = Boolean(currentSessionId) @@ -640,12 +641,15 @@ export const ChatContainer: React.FC = () => { if (!currentSessionId) return; if (hasSessionMessagesEntry && hasHistoryMetadata) return; + const isSessionSwitch = lastHydratedSessionRef.current !== currentSessionId; + lastHydratedSessionRef.current = currentSessionId; + const load = async () => { await loadMessages(currentSessionId).finally(() => { const statusType = sessionStatusForCurrent.type ?? 'idle'; const isActivePhase = statusType === 'busy' || statusType === 'retry'; const hasHashTarget = typeof window !== 'undefined' && window.location.hash.length > 0; - const shouldSkipScroll = (isActivePhase && isPinned) || hasHashTarget; + const shouldSkipScroll = hasHashTarget || (isActivePhase && isPinned && !isSessionSwitch); if (!shouldSkipScroll) { if (typeof window === 'undefined') { diff --git a/packages/ui/src/components/chat/MessageList.tsx b/packages/ui/src/components/chat/MessageList.tsx index ca5f5c09..1a9e525e 100644 --- a/packages/ui/src/components/chat/MessageList.tsx +++ b/packages/ui/src/components/chat/MessageList.tsx @@ -1692,17 +1692,10 @@ const MessageList = React.forwardRef(({ return false; } - const container = resolveScrollContainer(); - if (!container) { - return false; - } - const virtualizerBehavior = behavior === 'smooth' ? 'smooth' : 'auto'; historyVirtualizer.scrollToIndex(index, { align: 'start', behavior: virtualizerBehavior }); - const targetTop = Math.max(0, container.scrollTop - 50); - container.scrollTo({ top: targetTop, behavior }); return true; - }, [historyEntries.length, historyVirtualizer, resolveScrollContainer, shouldVirtualizeHistory]); + }, [historyEntries.length, historyVirtualizer, shouldVirtualizeHistory]); const scrollMessageElementIntoView = React.useCallback((messageId: string, behavior: ScrollBehavior = 'auto') => { const container = resolveScrollContainer(); diff --git a/packages/ui/src/components/chat/hooks/useChatTimelineController.ts b/packages/ui/src/components/chat/hooks/useChatTimelineController.ts index b1d7606d..15f226c0 100644 --- a/packages/ui/src/components/chat/hooks/useChatTimelineController.ts +++ b/packages/ui/src/components/chat/hooks/useChatTimelineController.ts @@ -14,19 +14,17 @@ import { import type { TurnHistorySignals } from '../lib/turns/historySignals'; import { getMemoryLimits, type SessionHistoryMeta } from '@/stores/types/sessionTypes'; -const waitForFrames = async (count = 1): Promise => { - if (typeof window === 'undefined') { - return; - } - for (let index = 0; index < count; index += 1) { - await new Promise((resolve) => { - window.requestAnimationFrame(() => resolve()); - }); - } -}; - type ViewportAnchor = { messageId: string; offsetTop: number }; +type PendingScrollRequest = { + sessionId: string; + kind: 'turn' | 'message'; + id: string; + behavior: ScrollBehavior; + turnId: string | null; + resolve: (value: boolean) => void; +}; + interface UseChatTimelineControllerOptions { sessionId: string | null; messages: ChatMessageEntry[]; @@ -100,6 +98,8 @@ export const useChatTimelineController = ({ const historyMetaRef = React.useRef(historyMeta); const previousTurnCountRef = React.useRef(turnWindowModel.turnCount); const initializedSessionRef = React.useRef(null); + const pendingRenderResolversRef = React.useRef void>>([]); + const pendingScrollRequestRef = React.useRef(null); const historySignals = React.useMemo(() => { const defaultLimit = getMemoryLimits().HISTORICAL_MESSAGES; @@ -189,10 +189,78 @@ export const useChatTimelineController = ({ previousTurnCountRef.current = nextTurnCount; }, [turnWindowModel.turnCount]); + const resolvePendingRenderWaiters = React.useCallback(() => { + const resolvers = pendingRenderResolversRef.current; + if (resolvers.length === 0) { + return; + } + pendingRenderResolversRef.current = []; + resolvers.forEach((resolve) => resolve()); + }, []); + + const waitForNextRenderCommit = React.useCallback((): Promise => { + return new Promise((resolve) => { + pendingRenderResolversRef.current.push(resolve); + }); + }, []); + + const resolvePendingScrollRequest = React.useCallback((value: boolean) => { + const pending = pendingScrollRequestRef.current; + if (!pending) { + return; + } + pendingScrollRequestRef.current = null; + pending.resolve(value); + }, []); + + const attemptPendingScrollRequest = React.useCallback(() => { + const pending = pendingScrollRequestRef.current; + if (!pending) { + return; + } + + if (pending.sessionId !== sessionIdRef.current) { + resolvePendingScrollRequest(false); + return; + } + + const didScroll = pending.kind === 'turn' + ? (messageListRef.current?.scrollToTurnId(pending.id, { behavior: pending.behavior }) ?? false) + : (messageListRef.current?.scrollToMessageId(pending.id, { behavior: pending.behavior }) ?? false); + + if (didScroll) { + if (pending.turnId) { + setActiveTurnId(pending.turnId); + } + resolvePendingScrollRequest(true); + return; + } + + const targetIndex = pending.kind === 'turn' + ? turnModelRef.current.turnIndexById.get(pending.id) + : turnModelRef.current.messageToTurnIndex.get(pending.id); + + if (typeof targetIndex === 'number' && targetIndex >= turnStartRef.current) { + resolvePendingScrollRequest(false); + } + }, [messageListRef, resolvePendingScrollRequest]); + + React.useEffect(() => { + return () => { + resolvePendingRenderWaiters(); + resolvePendingScrollRequest(false); + }; + }, [resolvePendingRenderWaiters, resolvePendingScrollRequest]); + const renderedMessages = React.useMemo(() => { return windowMessagesByTurn(messages, turnWindowModel, turnStart); }, [messages, turnStart, turnWindowModel]); + React.useLayoutEffect(() => { + resolvePendingRenderWaiters(); + attemptPendingScrollRequest(); + }, [attemptPendingScrollRequest, renderedMessages, resolvePendingRenderWaiters, turnStart]); + // --- Synchronous scroll compensation for load-more / reveal --- // fetchOlderHistory and revealBufferedTurns store a snapshot here // before triggering the state change. useLayoutEffect consumes it @@ -257,10 +325,10 @@ export const useChatTimelineController = ({ return next > 0 ? next : 0; }); - await waitForFrames(1); + await waitForNextRenderCommit(); setPendingRevealWork(false); return true; - }, [captureViewportAnchor, scrollRef]); + }, [captureViewportAnchor, scrollRef, waitForNextRenderCommit]); const fetchOlderHistory = React.useCallback(async (input: { preserveViewport: boolean; @@ -344,26 +412,29 @@ export const useChatTimelineController = ({ if (turnIndex < turnStartRef.current) { setTurnStart(turnIndex); - await waitForFrames(2); } - const didScroll = messageListRef.current?.scrollToTurnId(turnId, { - behavior: options?.behavior, - }) ?? false; + const result = await new Promise((resolve) => { + pendingScrollRequestRef.current = { + sessionId: sessionIdRef.current ?? sessionId ?? '', + kind: 'turn', + id: turnId, + behavior: options?.behavior ?? 'auto', + turnId, + resolve, + }; + attemptPendingScrollRequest(); + }); - if (didScroll) { - setActiveTurnId(turnId); + if (result) { return true; } - await waitForFrames(2); - return messageListRef.current?.scrollToTurnId(turnId, { - behavior: options?.behavior, - }) ?? false; + return false; } finally { setPendingRevealWork(false); } - }, [messageListRef, sessionId]); + }, [attemptPendingScrollRequest, sessionId]); const scrollToMessage = React.useCallback(async ( messageId: string, @@ -389,28 +460,29 @@ export const useChatTimelineController = ({ if (turnIndex < turnStartRef.current) { setTurnStart(turnIndex); - await waitForFrames(2); } - const didScroll = messageListRef.current?.scrollToMessageId(messageId, { - behavior: options?.behavior, - }) ?? false; + const result = await new Promise((resolve) => { + pendingScrollRequestRef.current = { + sessionId: sessionIdRef.current ?? sessionId ?? '', + kind: 'message', + id: messageId, + behavior: options?.behavior ?? 'auto', + turnId: turnId ?? null, + resolve, + }; + attemptPendingScrollRequest(); + }); - if (didScroll) { - if (turnId) { - setActiveTurnId(turnId); - } + if (result) { return true; } - await waitForFrames(2); - return messageListRef.current?.scrollToMessageId(messageId, { - behavior: options?.behavior, - }) ?? false; + return false; } finally { setPendingRevealWork(false); } - }, [messageListRef, sessionId]); + }, [attemptPendingScrollRequest, sessionId]); const resumeToBottom = React.useCallback(() => { const nextStart = getInitialTurnStart(turnModelRef.current.turnCount); diff --git a/packages/ui/src/hooks/useChatScrollManager.ts b/packages/ui/src/hooks/useChatScrollManager.ts index a17a5e2d..1f08d96c 100644 --- a/packages/ui/src/hooks/useChatScrollManager.ts +++ b/packages/ui/src/hooks/useChatScrollManager.ts @@ -114,6 +114,7 @@ export const useChatScrollManager = ({ const pinnedSyncRafRef = React.useRef(null); const preferInstantPinRef = React.useRef(false); const autoFollowDuringWorkRef = React.useRef(false); + const pendingSessionSwitchSnapRef = React.useRef(false); const viewportAnchorTimerRef = React.useRef | null>(null); const pendingViewportAnchorRef = React.useRef<{ sessionId: string; anchor: number } | null>(null); const lastViewportAnchorRef = React.useRef<{ sessionId: string; anchor: number } | null>(null); @@ -206,21 +207,35 @@ export const useChatScrollManager = ({ pinnedSyncRafRef.current = null; updateScrollButtonVisibility(); if (!isPinnedRef.current) { + pendingSessionSwitchSnapRef.current = false; return; } const distanceFromBottom = getDistanceFromBottom(); if (sessionIsWorking) { - if (distanceFromBottom <= 0.5) { + if (pendingSessionSwitchSnapRef.current && distanceFromBottom > 0.5) { + autoFollowDuringWorkRef.current = false; + scrollToBottomInternal({ instant: true }); + pendingSessionSwitchSnapRef.current = false; preferInstantPinRef.current = false; return; } + if (distanceFromBottom <= 0.5) { + if (!pendingSessionSwitchSnapRef.current) { + preferInstantPinRef.current = false; + } + return; + } + scrollPinnedToBottom(distanceFromBottom); preferInstantPinRef.current = false; + pendingSessionSwitchSnapRef.current = false; return; } + pendingSessionSwitchSnapRef.current = false; + if (distanceFromBottom <= getAutoFollowThreshold()) { preferInstantPinRef.current = false; return; @@ -480,6 +495,7 @@ export const useChatScrollManager = ({ flushViewportAnchor(); pendingViewportAnchorRef.current = null; autoFollowDuringWorkRef.current = false; + pendingSessionSwitchSnapRef.current = true; // Always start pinned at bottom on session switch preferInstantPinRef.current = true; @@ -497,6 +513,7 @@ export const useChatScrollManager = ({ React.useEffect(() => { if (!sessionIsWorking) { autoFollowDuringWorkRef.current = false; + pendingSessionSwitchSnapRef.current = false; } }, [sessionIsWorking]); diff --git a/packages/web/server/index.js b/packages/web/server/index.js index c1aabfd1..5c517146 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -407,7 +407,19 @@ const { const ENV_SKIP_OPENCODE_START = process.env.OPENCODE_SKIP_START === 'true' || process.env.OPENCHAMBER_SKIP_OPENCODE_START === 'true'; -const ENV_DESKTOP_NOTIFY = process.env.OPENCHAMBER_DESKTOP_NOTIFY === 'true'; +const ENV_DESKTOP_NOTIFY = (() => { + if (process.env.OPENCHAMBER_DESKTOP_NOTIFY === 'true') { + return true; + } + + if (process.env.OPENCHAMBER_RUNTIME === 'desktop') { + return true; + } + + const argv0 = typeof process.argv?.[0] === 'string' ? process.argv[0] : ''; + const argv1 = typeof process.argv?.[1] === 'string' ? process.argv[1] : ''; + return /openchamber-server/i.test(argv0) || /openchamber-server/i.test(argv1); +})(); const ENV_CONFIGURED_OPENCODE_WSL_DISTRO = typeof process.env.OPENCODE_WSL_DISTRO === 'string' && process.env.OPENCODE_WSL_DISTRO.trim().length > 0 ? process.env.OPENCODE_WSL_DISTRO.trim() @@ -758,6 +770,11 @@ const bootstrapOpenCodeAtStartup = async (...args) => { if (openCodeLifecycleState.openCodeProcess && !openCodeLifecycleState.isExternalOpenCode) { startHealthMonitoring(); } + if (ENV_DESKTOP_NOTIFY) { + void ensureGlobalWatcherStarted().catch((error) => { + console.warn(`Global event watcher startup failed: ${error?.message || error}`); + }); + } }; const killProcessOnPort = (...args) => openCodeLifecycleRuntime.killProcessOnPort(...args); @@ -874,6 +891,7 @@ async function main(options = {}) { opencodeWslDistro: resolvedWslDistro || null, nodeBinaryResolved: resolvedNodeBinary || null, bunBinaryResolved: resolvedBunBinary || null, + desktopNotifyEnabled: ENV_DESKTOP_NOTIFY, planModeExperimentalEnabled: PLAN_MODE_EXPERIMENT_ENABLED, }), uiPassword, @@ -891,6 +909,8 @@ async function main(options = {}) { removePushSubscription, updateUiVisibility, isUiVisible, + getUiNotificationClients: () => uiNotificationClients, + writeSseEvent, sessionRuntime, setPushInitialized, fs, diff --git a/packages/web/server/lib/notifications/routes.js b/packages/web/server/lib/notifications/routes.js index de550285..ceb4b06f 100644 --- a/packages/web/server/lib/notifications/routes.js +++ b/packages/web/server/lib/notifications/routes.js @@ -35,6 +35,8 @@ export const registerNotificationRoutes = (app, dependencies) => { removePushSubscription, updateUiVisibility, isUiVisible, + getUiNotificationClients, + writeSseEvent, getSessionActivitySnapshot, getSessionStateSnapshot, getSessionAttentionSnapshot, @@ -158,6 +160,35 @@ export const registerNotificationRoutes = (app, dependencies) => { }); }); + app.get('/api/notifications/stream', async (req, res) => { + const uiToken = uiAuthController?.ensureSessionToken + ? await uiAuthController.ensureSessionToken(req, res) + : getUiSessionTokenFromRequest(req); + if (!uiToken) { + return; + } + + res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); + res.setHeader('Cache-Control', 'no-cache, no-transform'); + res.setHeader('Connection', 'keep-alive'); + res.flushHeaders?.(); + + const clients = getUiNotificationClients(); + clients.add(res); + + try { + writeSseEvent(res, { + type: 'openchamber:notification-stream-ready', + properties: { uiToken }, + }); + } catch { + } + + req.on('close', () => { + clients.delete(res); + }); + }); + app.get('/api/session-activity', (_req, res) => { void ensureSessionWatcher(); res.json(getSessionActivitySnapshot()); diff --git a/packages/web/server/lib/opencode/bootstrap-runtime.js b/packages/web/server/lib/opencode/bootstrap-runtime.js index f70111d2..d1a91845 100644 --- a/packages/web/server/lib/opencode/bootstrap-runtime.js +++ b/packages/web/server/lib/opencode/bootstrap-runtime.js @@ -33,6 +33,8 @@ export const createBootstrapRuntime = (dependencies) => { removePushSubscription, updateUiVisibility, isUiVisible, + getUiNotificationClients, + writeSseEvent, sessionRuntime, setPushInitialized, fs, @@ -84,6 +86,8 @@ export const createBootstrapRuntime = (dependencies) => { removePushSubscription, updateUiVisibility, isUiVisible, + getUiNotificationClients, + writeSseEvent, getSessionActivitySnapshot: sessionRuntime.getSessionActivitySnapshot, getSessionStateSnapshot: sessionRuntime.getSessionStateSnapshot, getSessionAttentionSnapshot: sessionRuntime.getSessionAttentionSnapshot, diff --git a/packages/web/server/lib/opencode/watcher.js b/packages/web/server/lib/opencode/watcher.js index 1523c4e9..c8d20091 100644 --- a/packages/web/server/lib/opencode/watcher.js +++ b/packages/web/server/lib/opencode/watcher.js @@ -1,14 +1,27 @@ +import { createOpencodeClient } from '@opencode-ai/sdk/v2'; + export const createOpenCodeWatcherRuntime = (deps) => { const { waitForOpenCodePort, buildOpenCodeUrl, getOpenCodeAuthHeaders, - parseSseDataPayload, onPayload, } = deps; let abortController = null; + const unwrapGlobalEventPayload = (eventData) => { + if (!eventData || typeof eventData !== 'object') { + return null; + } + + if (eventData.payload && typeof eventData.payload === 'object') { + return eventData.payload; + } + + return eventData; + }; + const start = async () => { if (abortController) { return; @@ -23,62 +36,38 @@ export const createOpenCodeWatcherRuntime = (deps) => { const run = async () => { while (!signal.aborted) { attempt += 1; - let upstream; - let reader; try { - const url = buildOpenCodeUrl('/global/event', ''); - upstream = await fetch(url, { - headers: { - Accept: 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - ...getOpenCodeAuthHeaders(), - }, - signal, + const baseUrl = buildOpenCodeUrl('/', '').replace(/\/$/, ''); + const client = createOpencodeClient({ + baseUrl, + headers: getOpenCodeAuthHeaders(), }); - if (!upstream.ok || !upstream.body) { - throw new Error(`bad status ${upstream.status}`); - } + const result = await client.global.event({ + signal, + sseMaxRetryAttempts: 0, + onSseEvent: (event) => { + const payload = unwrapGlobalEventPayload(event.data); + if (!payload || typeof payload !== 'object') { + return; + } + onPayload(payload); + }, + }); console.log('[PushWatcher] connected'); - const decoder = new TextDecoder(); - reader = upstream.body.getReader(); - let buffer = ''; - - while (!signal.aborted) { - const { value, done } = await reader.read(); - if (done) { + for await (const _ of result.stream) { + void _; + if (signal.aborted) { 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); - separatorIndex = buffer.indexOf('\n\n'); - const payload = parseSseDataPayload(block); - onPayload(payload); - } } } catch (error) { if (signal.aborted) { return; } console.warn('[PushWatcher] disconnected', error?.message ?? error); - } finally { - try { - if (reader) { - await reader.cancel(); - reader.releaseLock(); - } else if (upstream?.body && !upstream.body.locked) { - await upstream.body.cancel(); - } - } catch { - } } const backoffMs = Math.min(1000 * Math.pow(2, Math.min(attempt, 5)), 30000);