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..3325857dd 100644 --- a/apps/desktop/src/app/desktop-controller.tsx +++ b/apps/desktop/src/app/desktop-controller.tsx @@ -15,9 +15,10 @@ 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 { storedSessionIdForNotification } from '../lib/session-ids' +import { isMessagingSource } from '../lib/session-source' import { latestSessionTodos } from '../lib/todos' import { setCronFocusJobId } from '../store/cron' import { @@ -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) @@ -164,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) @@ -364,6 +401,7 @@ export function DesktopController() { loadMoreSessions, loadMoreSessionsForProfile, refreshCronJobs, + refreshMessagingSessions, refreshSessions } = useSessionListActions({ profileScope }) @@ -512,6 +550,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 +924,58 @@ 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]) + + // 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' || !activeIsMessaging) { + 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) + } + }, [activeIsMessaging, gatewayState, refreshActiveMessagingTranscript]) + 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) 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