fix: migrate goals across compression sessions

agent/conversation_compression.py

@@ -644,10 +644,10 @@ def _release_lock() -> None:
                 # per-session lookup with no parent walk, so without this an
                 # active goal silently dies at the boundary (#33618).
                 try:
-                    from hermes_cli.goals import migrate_goal_to_session
-                    migrate_goal_to_session(old_session_id, agent.session_id, reason="compression")
+                    from hermes_cli.goals import migrate_goal_session
+                    migrate_goal_session(old_session_id, agent.session_id, db=agent._session_db)
                 except Exception as _goal_err:
-                    logger.debug("Could not migrate goal on compression: %s", _goal_err)
+                    logger.debug("GoalManager migration on compression failed: %s", _goal_err)
                 # Auto-number the title for the continuation session
                 if old_title:
                     try:

cli.py

@@ -8638,7 +8638,18 @@ def _manual_compress(self, cmd_original: str = ""):
                     getattr(self.agent, "session_id", None)
                     and self.agent.session_id != self.session_id
                 ):
+                    old_session_id = self.session_id
                     self.session_id = self.agent.session_id
+                    try:
+                        from hermes_cli.goals import migrate_goal_session
+
+                        migrate_goal_session(
+                            old_session_id,
+                            self.session_id,
+                            db=getattr(self.agent, "_session_db", None),
+                        )
+                    except Exception as goal_err:
+                        logger.debug("GoalManager migration on manual /compress failed: %s", goal_err)
                     self._pending_title = None
                     # Manual /compress replaces conversation_history with a new
                     # compressed handoff for the child session. Persist it from

gateway/run.py

@@ -3071,6 +3071,113 @@ def _session_key_for_source(self, source: SessionSource) -> str:
             profile=_profile,
         )
 
+    def _migrate_goal_for_compression_chain(
+        self,
+        old_session_id: str,
+        new_session_id: str,
+    ) -> bool:
+        """Move any live /goal row along a verified compression chain to its tip."""
+        old_session_id = str(old_session_id or "")
+        new_session_id = str(new_session_id or "")
+        if not old_session_id or not new_session_id or old_session_id == new_session_id:
+            return False
+        db = getattr(self, "_session_db", None)
+        try:
+            chain = db.get_compression_chain(old_session_id) if db is not None else [old_session_id]
+        except Exception:
+            logger.debug(
+                "goal migration: compression chain lookup failed for %s",
+                old_session_id,
+                exc_info=True,
+            )
+            chain = [old_session_id]
+        if not chain or chain[-1] != new_session_id:
+            return False
+        try:
+            from hermes_cli.goals import migrate_goal_session
+        except Exception:
+            logger.debug("goal migration: goals module unavailable", exc_info=True)
+            return False
+        migrated = False
+        for candidate_session_id in chain[:-1]:
+            try:
+                if migrate_goal_session(candidate_session_id, new_session_id, db=db):
+                    migrated = True
+                    break
+            except Exception:
+                logger.debug(
+                    "goal migration failed for compression edge %s -> %s",
+                    candidate_session_id,
+                    new_session_id,
+                    exc_info=True,
+                )
+        return migrated
+
+    def _handle_compression_session_switch(
+        self,
+        *,
+        session_key: str,
+        session_entry: Any,
+        old_session_id: str,
+        new_session_id: str,
+        source: Optional[SessionSource] = None,
+        reason: str,
+    ):
+        """Update routing after a session compression split and migrate /goal state."""
+        old_session_id = str(old_session_id or "")
+        new_session_id = str(new_session_id or "")
+        if not old_session_id or not new_session_id or old_session_id == new_session_id:
+            return session_entry
+
+        switched_entry = None
+        switch_session = getattr(self.session_store, "switch_session", None)
+        if callable(switch_session):
+            try:
+                candidate = switch_session(session_key, new_session_id)
+                if getattr(candidate, "session_id", None) == new_session_id:
+                    switched_entry = candidate
+            except Exception:
+                logger.debug("compression switch: session-store switch failed", exc_info=True)
+
+        original_entry = session_entry
+        if switched_entry is not None:
+            session_entry = switched_entry
+        elif session_entry is not None:
+            try:
+                session_entry.session_id = new_session_id
+            except Exception:
+                pass
+
+        # Keep direct-call tests and any already-held SessionEntry references in
+        # sync even when SessionStore.switch_session() returned a replacement.
+        if original_entry is not None and original_entry is not session_entry:
+            try:
+                original_entry.session_id = new_session_id
+            except Exception:
+                pass
+
+        store_entry = None
+        if switched_entry is None:
+            try:
+                entries = getattr(self.session_store, "_entries", None)
+                if isinstance(entries, dict) and session_key in entries:
+                    store_entry = entries[session_key]
+                    store_entry.session_id = new_session_id
+                    session_entry = store_entry
+            except Exception:
+                logger.debug("compression switch: session-store entry update failed", exc_info=True)
+
+            try:
+                self.session_store._save()
+            except Exception:
+                logger.debug("compression switch: session-store save failed", exc_info=True)
+
+        self._migrate_goal_for_compression_chain(old_session_id, new_session_id)
+
+        if source is not None and session_entry is not None:
+            self._sync_telegram_topic_binding(source, session_entry, reason=reason)
+        return session_entry
+
     def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool:
         """Return whether Telegram DM topic mode is active for this chat."""
         if source.platform != Platform.TELEGRAM or source.chat_type != "dm":
@@ -8846,7 +8953,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                 logger.debug("Failed to read Telegram topic binding", exc_info=True)
                 binding = None
             if binding:
-                bound_session_id = str(binding.get("session_id") or "")
+                binding_session_id = str(binding.get("session_id") or "")
+                bound_session_id = binding_session_id
                 # Heal bindings that point at a pre-compression parent: walk
                 # the compression-continuation chain forward to its tip so the
                 # next message resumes the compressed child instead of
@@ -8870,19 +8978,35 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     ):
                         bound_session_id = canonical_session_id
                 if bound_session_id and bound_session_id != session_entry.session_id:
-                    # Route the override through SessionStore so the session_key
-                    # → session_id mapping is persisted to disk and the previous
-                    # lane session is ended cleanly. Mutating session_entry in
-                    # place here created a split-brain state where the JSON
-                    # index pointed at one id but code downstream used another.
-                    switched = self.session_store.switch_session(session_key, bound_session_id)
-                    if switched is not None:
-                        session_entry = switched
-                # If the stored binding pointed at a parent, rewrite it to the
-                # canonical descendant now that we've followed the chain.
+                    if bound_session_id != binding_session_id:
+                        # Route compression-tip repairs through the compression
+                        # switch path so session-store persistence, topic-binding
+                        # sync, and /goal state migration cannot drift apart.
+                        session_entry = self._handle_compression_session_switch(
+                            session_key=session_key,
+                            session_entry=session_entry,
+                            old_session_id=binding_session_id,
+                            new_session_id=bound_session_id,
+                            source=source,
+                            reason="compression-tip-walk",
+                        )
+                    else:
+                        # Ordinary/restored topic bindings are not compression
+                        # migrations. Preserve the normal session-store switch
+                        # semantics so the static topic lane resumes the bound
+                        # session even when get_compression_tip() returns the
+                        # same ID.
+                        switched = self.session_store.switch_session(
+                            session_key, bound_session_id,
+                        )
+                        if switched is not None:
+                            session_entry = switched
+                # If the stored binding pointed at a parent but this lane was
+                # already on the canonical descendant, rewrite the binding now.
                 if (
                     bound_session_id
-                    and bound_session_id != str(binding.get("session_id") or "")
+                    and bound_session_id != binding_session_id
+                    and bound_session_id == session_entry.session_id
                 ):
                     self._sync_telegram_topic_binding(
                         source, session_entry, reason="compression-tip-walk",
@@ -9277,10 +9401,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                                         getattr(_hyg_agent, "compression_in_place", False)
                                     )
                                     if _hyg_rotated:
-                                        session_entry.session_id = _hyg_new_sid
-                                        self.session_store._save()
-                                        self._sync_telegram_topic_binding(
-                                            source, session_entry,
+                                        session_entry = self._handle_compression_session_switch(
+                                            session_key=session_key,
+                                            session_entry=session_entry,
+                                            old_session_id=session_entry.session_id,
+                                            new_session_id=_hyg_new_sid,
+                                            source=source,
                                             reason="hygiene-compression",
                                         )
 
@@ -9665,10 +9791,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
             # If the agent's session_id changed during compression, update
             # session_entry so transcript writes below go to the right session.
             if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id:
-                session_entry.session_id = agent_result["session_id"]
-                self.session_store._save()
-                self._sync_telegram_topic_binding(
-                    source, session_entry, reason="agent-result-compression",
+                session_entry = self._handle_compression_session_switch(
+                    session_key=session_key,
+                    session_entry=session_entry,
+                    old_session_id=session_entry.session_id,
+                    new_session_id=agent_result["session_id"],
+                    source=source,
+                    reason="agent-result-compression",
                 )
 
             # Prepend reasoning/thinking if display is enabled (per-platform).
@@ -11327,6 +11456,7 @@ def run_sync():
 
 
 
+
     async def _get_telegram_topic_capabilities(self, source: SessionSource) -> dict:
         """Read Telegram private-topic capability flags via Bot API getMe."""
         adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
@@ -16017,8 +16147,14 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 )
                 entry = self.session_store._entries.get(session_key)
                 if entry:
-                    entry.session_id = agent_session_id
-                    self.session_store._save()
+                    entry = self._handle_compression_session_switch(
+                        session_key=session_key,
+                        session_entry=entry,
+                        old_session_id=session_id,
+                        new_session_id=agent_session_id,
+                        source=source,
+                        reason="agent-run-compression",
+                    )
 
                 # If this is a Telegram DM and source.thread_id was lost during
                 # the session split (synthetic / recovered event), restore it
@@ -16050,10 +16186,6 @@ def _approval_notify_sync(approval_data: dict) -> None:
                             "Failed to restore thread_id from binding after session split",
                             exc_info=True,
                         )
-                if entry:
-                    self._sync_telegram_topic_binding(
-                        source, entry, reason="agent-run-compression",
-                    )
 
             effective_session_id = agent_session_id
             # history_offset=0 whenever the agent's message list no longer has

gateway/slash_commands.py

@@ -2701,10 +2701,13 @@ async def _handle_compress_command(self, event: MessageEvent) -> str:
                 rotated = new_session_id != session_entry.session_id
                 _in_place = bool(getattr(tmp_agent, "compression_in_place", False))
                 if rotated:
-                    session_entry.session_id = new_session_id
-                    self.session_store._save()
-                    self._sync_telegram_topic_binding(
-                        source, session_entry, reason="compress-command",
+                    session_entry = self._handle_compression_session_switch(
+                        session_key=session_key,
+                        session_entry=session_entry,
+                        old_session_id=session_entry.session_id,
+                        new_session_id=new_session_id,
+                        source=source,
+                        reason="compress-command",
                     )
 
                 # Rewrite the transcript when EITHER rotation produced a new id