From adcc450e803abaa5911e74aef02d97b2e8f2861e Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Thu, 29 Jan 2026 00:40:59 +0200 Subject: [PATCH] feat: add message stall detection and soft resync in event stream --- packages/ui/src/hooks/useEventStream.ts | 167 +++++++++++++++++------- 1 file changed, 123 insertions(+), 44 deletions(-) diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 344afc57..dfe0bcae 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -22,6 +22,16 @@ interface EventData { properties?: Record; } +const readStringProp = (obj: unknown, keys: string[]): string | null => { + if (!obj || typeof obj !== 'object') return null; + const record = obj as Record; + for (let i = 0; i < keys.length; i++) { + const value = record[keys[i]]; + if (typeof value === 'string' && value.length > 0) return value; + } + return null; +}; + type MessageTracker = (messageId: string, event?: string, extraData?: Record) => void; declare global { @@ -347,6 +357,10 @@ export const useEventStream = () => { [resyncMessages] ); + React.useEffect(() => { + scheduleSoftResyncRef.current = scheduleSoftResync; + }, [scheduleSoftResync]); + const trackMessage = React.useCallback((messageId: string, event?: string, extraData?: Record) => { if (streamDebugEnabled()) { console.debug(`[MessageTracker] ${messageId}: ${event}`, extraData); @@ -389,6 +403,14 @@ export const useEventStream = () => { const pauseTimeoutRef = React.useRef(null); const staleCheckIntervalRef = React.useRef(null); const lastEventTimestampRef = React.useRef(Date.now()); + const lastMessageEventBySessionRef = React.useRef>(new Map()); + const pendingMessageStallTimersRef = React.useRef>(new Map()); + const lastMessageStallRecoveryBySessionRef = React.useRef>(new Map()); + + const scheduleSoftResyncRef = React.useRef< + (sessionId: string, reason: string, limit?: number) => Promise + >(() => Promise.resolve()); + const scheduleReconnectRef = React.useRef<(hint?: string) => void>(() => {}); const isDesktopRuntimeRef = React.useRef(false); const maybeBootstrapIfStale = React.useCallback( @@ -493,6 +515,52 @@ export const useEventStream = () => { return; } + const shouldArmMessageStallCheck = storePhase === 'idle' && (phase === 'busy' || phase === 'cooldown'); + const shouldDisarmMessageStallCheck = phase === 'idle'; + + if (shouldDisarmMessageStallCheck) { + const pending = pendingMessageStallTimersRef.current.get(sessionId); + if (pending) { + clearTimeout(pending); + pendingMessageStallTimersRef.current.delete(sessionId); + } + } + + if (shouldArmMessageStallCheck) { + const pending = pendingMessageStallTimersRef.current.get(sessionId); + if (pending) { + clearTimeout(pending); + pendingMessageStallTimersRef.current.delete(sessionId); + } + + const startAt = Date.now(); + const timer = setTimeout(() => { + const currentPhase = useSessionStore.getState().sessionActivityPhase?.get(sessionId); + if (currentPhase !== 'busy' && currentPhase !== 'cooldown') { + return; + } + + const lastRecoveryAt = lastMessageStallRecoveryBySessionRef.current.get(sessionId) ?? 0; + if (Date.now() - lastRecoveryAt < 15000) { + return; + } + + const lastMsgAt = lastMessageEventBySessionRef.current.get(sessionId) ?? 0; + if (lastMsgAt >= startAt) { + return; + } + + lastMessageStallRecoveryBySessionRef.current.set(sessionId, Date.now()); + + void scheduleSoftResyncRef.current(sessionId, 'activity_started_no_message', getActiveSessionWindow()) + .finally(() => { + scheduleReconnectRef.current('No message events after activity start'); + }); + }, 2000); + + pendingMessageStallTimersRef.current.set(sessionId, timer); + } + const existingTimer = sessionCooldownTimersRef.current.get(sessionId); if (existingTimer) { clearTimeout(existingTimer); @@ -737,25 +805,19 @@ export const useEventStream = () => { const partExt = part as Record; const messageInfo = (typeof props.info === 'object' && props.info !== null) ? (props.info as Record) : props; - const messageInfoSessionId = typeof (messageInfo as { sessionID?: unknown }).sessionID === 'string' - ? (messageInfo as { sessionID?: string }).sessionID - : null; + const messageInfoSessionId = readStringProp(messageInfo, ['sessionID', 'sessionId']); const resolvedSessionId = - (typeof partExt.sessionID === 'string' && (partExt.sessionID as string).length > 0) ? (partExt.sessionID as string) : - (typeof messageInfoSessionId === 'string' && messageInfoSessionId.length > 0) ? messageInfoSessionId : - (typeof props.sessionID === 'string' && (props.sessionID as string).length > 0) ? (props.sessionID as string) : - null; + readStringProp(partExt, ['sessionID', 'sessionId']) || + messageInfoSessionId || + readStringProp(props, ['sessionID', 'sessionId']); - const messageInfoId = typeof (messageInfo as { id?: unknown }).id === 'string' - ? (messageInfo as { id?: string }).id - : null; + const messageInfoId = readStringProp(messageInfo, ['messageID', 'messageId', 'id']); const resolvedMessageId = - (typeof partExt.messageID === 'string' && (partExt.messageID as string).length > 0) ? (partExt.messageID as string) : - (typeof messageInfoId === 'string' && messageInfoId.length > 0) ? messageInfoId : - (typeof props.messageID === 'string' && (props.messageID as string).length > 0) ? (props.messageID as string) : - null; + readStringProp(partExt, ['messageID', 'messageId']) || + messageInfoId || + readStringProp(props, ['messageID', 'messageId']); if (!resolvedSessionId || !resolvedMessageId) { if (streamDebugEnabled()) { @@ -770,6 +832,13 @@ export const useEventStream = () => { const sessionId = resolvedSessionId; const messageId = resolvedMessageId; + lastMessageEventBySessionRef.current.set(sessionId, Date.now()); + const pendingTimer = pendingMessageStallTimersRef.current.get(sessionId); + if (pendingTimer) { + clearTimeout(pendingTimer); + pendingMessageStallTimersRef.current.delete(sessionId); + } + const trimmedHeadMaxId = useSessionStore.getState().sessionMemoryState.get(sessionId)?.trimmedHeadMaxId; if (trimmedHeadMaxId && !isIdNewer(messageId, trimmedHeadMaxId)) { if (streamDebugEnabled()) { @@ -817,14 +886,12 @@ export const useEventStream = () => { const messageExt = message as Record; const resolvedSessionId = - (typeof messageExt.sessionID === 'string' && (messageExt.sessionID as string).length > 0) ? (messageExt.sessionID as string) : - (typeof props.sessionID === 'string' && (props.sessionID as string).length > 0) ? (props.sessionID as string) : - null; + readStringProp(messageExt, ['sessionID', 'sessionId']) || + readStringProp(props, ['sessionID', 'sessionId']); const resolvedMessageId = - (typeof messageExt.id === 'string' && (messageExt.id as string).length > 0) ? (messageExt.id as string) : - (typeof props.messageID === 'string' && (props.messageID as string).length > 0) ? (props.messageID as string) : - null; + readStringProp(messageExt, ['messageID', 'messageId', 'id']) || + readStringProp(props, ['messageID', 'messageId']); if (!resolvedSessionId || !resolvedMessageId) { if (streamDebugEnabled()) { @@ -839,6 +906,13 @@ export const useEventStream = () => { const sessionId = resolvedSessionId; const messageId = resolvedMessageId; + lastMessageEventBySessionRef.current.set(sessionId, Date.now()); + const pendingTimer = pendingMessageStallTimersRef.current.get(sessionId); + if (pendingTimer) { + clearTimeout(pendingTimer); + pendingMessageStallTimersRef.current.delete(sessionId); + } + const trimmedHeadMaxId = useSessionStore.getState().sessionMemoryState.get(sessionId)?.trimmedHeadMaxId; if (trimmedHeadMaxId && !isIdNewer(messageId, trimmedHeadMaxId)) { if (streamDebugEnabled()) { @@ -1776,6 +1850,10 @@ export const useEventStream = () => { }, delay); }, [shouldHoldConnection, stopStream, publishStatus, startStream]); + React.useEffect(() => { + scheduleReconnectRef.current = scheduleReconnect; + }, [scheduleReconnect]); + React.useEffect(() => { const cooldownTimers = sessionCooldownTimersRef.current; @@ -1923,32 +2001,33 @@ export const useEventStream = () => { clearInterval(staleCheckIntervalRef.current); } - staleCheckIntervalRef.current = setInterval(() => { - if (!shouldHoldConnection()) return; + staleCheckIntervalRef.current = setInterval(() => { + if (!shouldHoldConnection()) return; - const now = Date.now(); - const hasBusySessions = Array.from(useSessionStore.getState().sessionActivityPhase?.values?.() ?? []).some( - (phase) => phase === 'busy' || phase === 'cooldown' - ); - if (hasBusySessions) { - void refreshSessionActivityStatus(); - } - if (now - lastEventTimestampRef.current > 45000) { - Promise.resolve().then(async () => { - try { - const healthy = await opencodeClient.checkHealth(); - if (!healthy) { - scheduleReconnect('Refreshing stalled stream'); - } else { - lastEventTimestampRef.current = Date.now(); - } - } catch (error) { - console.warn('Health check after stale stream failed:', error); - scheduleReconnect('Refreshing stalled stream'); + const now = Date.now(); + const hasBusySessions = Array.from(useSessionStore.getState().sessionActivityPhase?.values?.() ?? []).some( + (phase) => phase === 'busy' || phase === 'cooldown' + ); + + if (hasBusySessions) { + void refreshSessionActivityStatus(); } - }); - } - }, 10000); + if (now - lastEventTimestampRef.current > 45000) { + Promise.resolve().then(async () => { + try { + const healthy = await opencodeClient.checkHealth(); + if (!healthy) { + scheduleReconnect('Refreshing stalled stream'); + } else { + lastEventTimestampRef.current = Date.now(); + } + } catch (error) { + console.warn('Health check after stale stream failed:', error); + scheduleReconnect('Refreshing stalled stream'); + } + }); + } + }, 10000); return () => { clearTimeout(startTimer);