From 52d0d671e79ec1d5727881a6e8519332d04c1a07 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 3 Jul 2026 04:29:22 -0500 Subject: [PATCH 1/3] fix(desktop): poll messaging sessions so platform traffic appears live Inbound Telegram/WeChat/Discord messages are written by the background gateway, not the desktop websocket that drives local chats. Without explicit polling the messaging sidebar and the open transcript stay frozen until the user manually refreshes. Desktop: - MESSAGING_POLL_INTERVAL_MS (10 s): interval poll of the messaging session list so new platform sessions surface automatically. - ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS (5 s): poll the currently- viewed messaging transcript and re-hydrate the chat state when the FNV-1a signature changes (hash covers role + timestamp + content). - sameCronSignature now compares lineage_root_id / source / profile / preview / message_count / last_active / ended_at so stale previews and activity times are no longer silently ignored. - sessionMatchesStoredId helper de-dups the id / _lineage_root_id check. - refreshMessagingSessions exposed from useSessionListActions so the controller can use it in the poll effect. Gateway: - SessionStore._compression_tip_for_session_id: look up the latest compression continuation for a session id. - SessionStore._heal_compression_tip_locked: rewrite a stale entry to the compression child before returning it, so a restart or failed send no longer leaves the store pinned to the compressed parent. Co-authored-by: lawyer112 --- .../src/app/desktop-controller-utils.ts | 17 ++- apps/desktop/src/app/desktop-controller.tsx | 120 +++++++++++++++++- .../session/hooks/use-session-list-actions.ts | 1 + gateway/session.py | 60 +++++++++ tests/gateway/test_restart_resume_pending.py | 21 +++ 5 files changed, 217 insertions(+), 2 deletions(-) diff --git a/apps/desktop/src/app/desktop-controller-utils.ts b/apps/desktop/src/app/desktop-controller-utils.ts index cb7d618e6..5754d69ef 100644 --- a/apps/desktop/src/app/desktop-controller-utils.ts +++ b/apps/desktop/src/app/desktop-controller-utils.ts @@ -7,5 +7,20 @@ export function sameCronSignature(a: SessionInfo[], b: SessionInfo[]): boolean { return false } - return a.every((session, i) => session.id === b[i]?.id && session.title === b[i]?.title) + return a.every((session, i) => { + const other = b[i] + + return ( + other != null && + session.id === other.id && + session._lineage_root_id === other._lineage_root_id && + session.title === other.title && + session.source === other.source && + session.profile === other.profile && + session.preview === other.preview && + session.message_count === other.message_count && + session.last_active === other.last_active && + session.ended_at === other.ended_at + ) + }) } diff --git a/apps/desktop/src/app/desktop-controller.tsx b/apps/desktop/src/app/desktop-controller.tsx index f3a07183e..ca78a9ccb 100644 --- a/apps/desktop/src/app/desktop-controller.tsx +++ b/apps/desktop/src/app/desktop-controller.tsx @@ -15,8 +15,9 @@ import { cn } from '@/lib/utils' import { useSkinCommand } from '@/themes/use-skin-command' import { formatRefValue } from '../components/assistant-ui/directive-text' -import { getSessionMessages, triggerCronJob } from '../hermes' +import { getSessionMessages, type SessionMessage, triggerCronJob } from '../hermes' import { type ChatMessage, chatMessageText, preserveLocalAssistantErrors, toChatMessages } from '../lib/chat-messages' +import { isMessagingSource } from '../lib/session-source' import { storedSessionIdForNotification } from '../lib/session-ids' import { latestSessionTodos } from '../lib/todos' import { setCronFocusJobId } from '../store/cron' @@ -60,6 +61,7 @@ import { $freshDraftReady, $gatewayState, $messages, + $messagingSessions, $resumeExhaustedSessionId, $resumeFailedSessionId, $selectedStoredSessionId, @@ -146,6 +148,39 @@ const SkillsView = lazy(async () => ({ default: (await import('./skills')).Skill // this cadence while the app is open + visible so new runs surface promptly // instead of waiting for the next user-triggered refreshSessions(). const CRON_POLL_INTERVAL_MS = 30_000 +// Messaging-platform turns are written by the background gateway (WeChat, +// Telegram, Discord, …), not the desktop websocket that drives local chats. +// Poll the bounded messaging slice while visible so inbound platform traffic +// appears without requiring a manual refresh or route change. +const MESSAGING_POLL_INTERVAL_MS = 10_000 +const ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS = 5_000 + +function sessionMatchesStoredId(session: { id: string; _lineage_root_id?: null | string }, id: string): boolean { + return session.id === id || session._lineage_root_id === id +} + +function hashString(hash: number, value: string): number { + let next = hash + + for (let i = 0; i < value.length; i++) { + next ^= value.charCodeAt(i) + next = Math.imul(next, 16777619) + } + + return next >>> 0 +} + +function sessionMessagesSignature(messages: SessionMessage[]): string { + let hash = 2166136261 + + for (const m of messages) { + hash = hashString(hash, m.role) + hash = hashString(hash, String(m.timestamp ?? '')) + hash = hashString(hash, typeof m.content === 'string' ? m.content : JSON.stringify(m.content) ?? '') + } + + return `${messages.length}:${hash}` +} export function DesktopController() { const queryClient = useQueryClient() @@ -154,6 +189,7 @@ export function DesktopController() { const busyRef = useRef(false) const creatingSessionRef = useRef(false) + const messagingTranscriptSignatureRef = useRef(new Map()) const gatewayState = useStore($gatewayState) const activeSessionId = useStore($activeSessionId) @@ -364,6 +400,7 @@ export function DesktopController() { loadMoreSessions, loadMoreSessionsForProfile, refreshCronJobs, + refreshMessagingSessions, refreshSessions } = useSessionListActions({ profileScope }) @@ -512,6 +549,42 @@ export function DesktopController() { [activeSessionIdRef, selectedStoredSessionIdRef, updateSessionState] ) + const refreshActiveMessagingTranscript = useCallback(async () => { + const storedSessionId = selectedStoredSessionIdRef.current + const runtimeSessionId = activeSessionIdRef.current + + if (!storedSessionId || !runtimeSessionId || busyRef.current) { + return + } + + const stored = $messagingSessions.get().find(s => sessionMatchesStoredId(s, storedSessionId)) + + if (!stored || !isMessagingSource(stored.source)) { + return + } + + try { + const latest = await getSessionMessages(storedSessionId, stored.profile) + const signatureKey = `${stored.profile ?? 'default'}:${storedSessionId}` + const sig = sessionMessagesSignature(latest.messages) + + if (messagingTranscriptSignatureRef.current.get(signatureKey) === sig) { + return + } + + messagingTranscriptSignatureRef.current.set(signatureKey, sig) + const messages = toChatMessages(latest.messages) + + updateSessionState( + runtimeSessionId, + state => ({ ...state, messages: preserveLocalAssistantErrors(messages, state.messages) }), + storedSessionId + ) + } catch { + // Non-fatal: next poll or manual refresh can hydrate. + } + }, [activeSessionIdRef, busyRef, selectedStoredSessionIdRef, updateSessionState]) + const { handleGatewayEvent } = useMessageStream({ activeSessionIdRef, hydrateFromStoredSession, @@ -850,6 +923,51 @@ export function DesktopController() { } }, [gatewayState, refreshCronJobs]) + // Keep messaging-platform session lists live: inbound Telegram/WeChat/Discord + // turns are written by the gateway, not the desktop websocket, so they won't + // appear without polling. + useEffect(() => { + if (gatewayState !== 'open') { + return + } + + const tick = () => { + if (document.visibilityState === 'visible') { + void refreshMessagingSessions() + } + } + + const intervalId = window.setInterval(tick, MESSAGING_POLL_INTERVAL_MS) + document.addEventListener('visibilitychange', tick) + + return () => { + window.clearInterval(intervalId) + document.removeEventListener('visibilitychange', tick) + } + }, [gatewayState, refreshMessagingSessions]) + + // Keep the currently-viewed messaging transcript live. + useEffect(() => { + if (gatewayState !== 'open' || !selectedStoredSessionId) { + return + } + + const tick = () => { + if (document.visibilityState === 'visible') { + void refreshActiveMessagingTranscript() + } + } + + const intervalId = window.setInterval(tick, ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS) + document.addEventListener('visibilitychange', tick) + tick() + + return () => { + window.clearInterval(intervalId) + document.removeEventListener('visibilitychange', tick) + } + }, [gatewayState, refreshActiveMessagingTranscript, selectedStoredSessionId]) + useEffect(() => { if (gatewayState === 'open' && !activeSessionId && freshDraftReady) { void refreshCurrentModel() diff --git a/apps/desktop/src/app/session/hooks/use-session-list-actions.ts b/apps/desktop/src/app/session/hooks/use-session-list-actions.ts index 6c5d89e71..6f971c316 100644 --- a/apps/desktop/src/app/session/hooks/use-session-list-actions.ts +++ b/apps/desktop/src/app/session/hooks/use-session-list-actions.ts @@ -225,6 +225,7 @@ export function useSessionListActions({ profileScope }: UseSessionListActionsArg loadMoreSessions, loadMoreSessionsForProfile, refreshCronJobs, + refreshMessagingSessions, refreshSessions } } diff --git a/gateway/session.py b/gateway/session.py index 67bd84aaa..5ced2d8aa 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1369,6 +1369,48 @@ class SessionStore: return None + def _compression_tip_for_session_id(self, session_id: Optional[str]) -> Optional[str]: + """Return the latest compression continuation for *session_id*. + + When an agent compresses context mid-turn the transcript moves to a + child session, but a restart or failed send can leave the SessionStore + mapping pointing at the compressed parent. Heal that on read so the + next inbound message resumes the child instead of reloading the parent. + """ + if not session_id or self._db is None: + return session_id + try: + return self._db.get_compression_tip(session_id) or session_id + except Exception: + logger.debug( + "Compression-tip lookup failed for session %s", + session_id, + exc_info=True, + ) + return session_id + + def _heal_compression_tip_locked( + self, + entry: "SessionEntry", + original_session_id: Optional[str], + canonical_session_id: Optional[str], + ) -> bool: + """Rewrite *entry* to the compression continuation if stale. Lock held.""" + if ( + not original_session_id + or not canonical_session_id + or entry.session_id != original_session_id + or canonical_session_id == original_session_id + ): + return False + logger.info( + "SessionStore healed compressed session mapping: %s -> %s", + entry.session_id, + canonical_session_id, + ) + entry.session_id = canonical_session_id + return True + def has_any_sessions(self) -> bool: """Check if any sessions have ever been created (across all platforms). @@ -1409,12 +1451,30 @@ class SessionStore: # All _entries / _loaded mutations are protected by self._lock. db_end_session_id = None db_create_kwargs = None + existing_session_id = None + + if not force_new: + with self._lock: + self._ensure_loaded_locked() + entry = self._entries.get(session_key) + if entry is not None: + existing_session_id = entry.session_id + + # Look up the compression continuation outside the lock (DB I/O). + canonical_existing_session_id = ( + self._compression_tip_for_session_id(existing_session_id) + if existing_session_id + else None + ) with self._lock: self._ensure_loaded_locked() if session_key in self._entries and not force_new: entry = self._entries[session_key] + self._heal_compression_tip_locked( + entry, existing_session_id, canonical_existing_session_id + ) # Self-heal stale routing: if this session_key still points at # a session that has ALREADY been ended in state.db (end_reason diff --git a/tests/gateway/test_restart_resume_pending.py b/tests/gateway/test_restart_resume_pending.py index 9b159ae3b..23c5e674f 100644 --- a/tests/gateway/test_restart_resume_pending.py +++ b/tests/gateway/test_restart_resume_pending.py @@ -384,6 +384,27 @@ class TestGetOrCreateResumePending: # Flag is NOT cleared on read — only on successful turn completion. assert second.resume_pending is True + def test_resume_pending_follows_compression_tip(self, tmp_path): + """Interrupted platform mappings must not stay pinned to compressed roots.""" + store = _make_store(tmp_path) + source = _make_source( + platform=Platform.WEIXIN, + chat_id="wx-chat", + user_id="wx-user", + ) + first = store.get_or_create_session(source) + original_sid = first.session_id + store.mark_resume_pending(first.session_key) + + with patch.object( + store, "_compression_tip_for_session_id", return_value="child-session" + ) as mock_tip: + second = store.get_or_create_session(source) + + assert second.session_id == "child-session" + assert second.resume_pending is True + mock_tip.assert_called_with(original_sid) + def test_suspended_still_creates_new_session(self, tmp_path): """Regression guard — suspended must still force a clean slate.""" store = _make_store(tmp_path) From dfb28cc6315bafa3a9a08714be2ab95b90d1beab Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 3 Jul 2026 04:38:45 -0500 Subject: [PATCH 2/3] refactor(desktop): only poll the transcript when it's a messaging session The active-transcript poll armed a 5 s timer for every selected session and no-op'd inside the tick for local chats (already live over the websocket). Derive activeIsMessaging and gate the effect on it so local chats never spin an idle timer. --- apps/desktop/src/app/desktop-controller.tsx | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/apps/desktop/src/app/desktop-controller.tsx b/apps/desktop/src/app/desktop-controller.tsx index ca78a9ccb..3325857dd 100644 --- a/apps/desktop/src/app/desktop-controller.tsx +++ b/apps/desktop/src/app/desktop-controller.tsx @@ -17,8 +17,8 @@ import { useSkinCommand } from '@/themes/use-skin-command' import { formatRefValue } from '../components/assistant-ui/directive-text' import { getSessionMessages, type SessionMessage, triggerCronJob } from '../hermes' import { type ChatMessage, chatMessageText, preserveLocalAssistantErrors, toChatMessages } from '../lib/chat-messages' -import { isMessagingSource } from '../lib/session-source' import { storedSessionIdForNotification } from '../lib/session-ids' +import { isMessagingSource } from '../lib/session-source' import { latestSessionTodos } from '../lib/todos' import { setCronFocusJobId } from '../store/cron' import { @@ -200,6 +200,7 @@ export function DesktopController() { const filePreviewTarget = useStore($filePreviewTarget) const previewTarget = useStore($previewTarget) const selectedStoredSessionId = useStore($selectedStoredSessionId) + const messagingSessions = useStore($messagingSessions) const terminalTakeover = useStore($terminalTakeover) const reviewOpen = useStore($reviewOpen) const fileBrowserOpen = useStore($fileBrowserOpen) @@ -946,9 +947,16 @@ export function DesktopController() { } }, [gatewayState, refreshMessagingSessions]) + // Only the open messaging transcript needs a poll — local chats are already + // live over the websocket, so arming a timer for them would just no-op every + // tick. Gate on the active session actually being a messaging source. + const activeIsMessaging = + !!selectedStoredSessionId && + isMessagingSource(messagingSessions.find(s => sessionMatchesStoredId(s, selectedStoredSessionId))?.source) + // Keep the currently-viewed messaging transcript live. useEffect(() => { - if (gatewayState !== 'open' || !selectedStoredSessionId) { + if (gatewayState !== 'open' || !activeIsMessaging) { return } @@ -966,7 +974,7 @@ export function DesktopController() { window.clearInterval(intervalId) document.removeEventListener('visibilitychange', tick) } - }, [gatewayState, refreshActiveMessagingTranscript, selectedStoredSessionId]) + }, [activeIsMessaging, gatewayState, refreshActiveMessagingTranscript]) useEffect(() => { if (gatewayState === 'open' && !activeSessionId && freshDraftReady) { From c1e825399cbd289d167b1e157eaaad6333c7fced Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 3 Jul 2026 04:46:01 -0500 Subject: [PATCH 3/3] test(gateway): stub get_compression_tip in stale-guard db mock The routing-heal added to get_or_create_session calls SessionDB.get_compression_tip; the stale-guard suite's bare MagicMock db returned a Mock the heal then assigned as session_id, failing JSON serialization. Model the real contract (a non-compressed session's tip is itself) so the heal is a correct no-op. --- tests/gateway/test_session_store_runtime_stale_guard.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/gateway/test_session_store_runtime_stale_guard.py b/tests/gateway/test_session_store_runtime_stale_guard.py index 262d1a489..57f8c624b 100644 --- a/tests/gateway/test_session_store_runtime_stale_guard.py +++ b/tests/gateway/test_session_store_runtime_stale_guard.py @@ -49,6 +49,10 @@ def _db_returning(rows: dict) -> MagicMock: db.find_latest_gateway_session_for_peer.return_value = None db.reopen_session.return_value = None db.create_session.return_value = None + # No compression continuation → the tip is the session itself (identity), + # mirroring the real SessionDB.get_compression_tip. Without this a bare Mock + # would return a Mock the routing heal then assigns as session_id. + db.get_compression_tip.side_effect = lambda sid: sid return db