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 <lawyer112@users.noreply.github.com>

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
+    )
+  })
 }

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<string, string>())
 
   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()

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
   }
 }

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

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)