fix(acp): also preserve archived rows on model-switch / restore saves

Follow-up widening the archived-history fix to the sibling save paths the
original PR did not cover. Model switches (_cmd_model, set_session_model) and
_restore mint a fresh AIAgent with _session_db_created=False, so the
agent-owns-persistence guard evaluates False and the blind full-history
replace_messages() fired — DELETEing the durable active=0/compacted=1 rows on
any compressed ACP session (same data-loss class the PR fixes, different
trigger).

- hermes_state.replace_messages: add active_only=True to delete/reinsert only
  the live (active=1) rows, leaving soft-archived rows untouched (idea adopted
  from the competing PR #50306 by @mrparker0980, credited).
- hermes_state.has_archived_messages: cheap existence probe for active=0 rows.
- acp_adapter._persist: when the agent doesn't own persistence but the session
  already has archived rows on disk, replace active-only; otherwise the
  destructive full replace stays (fresh create/fork has nothing to lose).
- Regression test: model-switch save on a compacted session keeps the archived
  turn discoverable via get_messages(include_inactive=True) + search_messages.

acp_adapter/session.py

@@ -485,7 +485,23 @@ class SessionManager:
                 and bool(getattr(agent, "_session_db_created", False))
             )
             if not agent_owns_persistence:
-                db.replace_messages(state.session_id, state.history)
+                # Even when the current agent doesn't "own" persistence, the
+                # session on disk may already carry compaction-archived rows —
+                # e.g. after a model switch or a /restore, both of which mint a
+                # fresh agent with _session_db_created=False (so the check above
+                # is False) yet leave the durable archived transcript in place.
+                # A full-history replace would DELETE those archived rows just
+                # like the owned-agent case. Guard against it: when archived
+                # rows exist, replace ONLY the live (active=1) set and leave the
+                # archived turns untouched; otherwise the destructive replace is
+                # safe (fresh create/fork with no archived history to lose).
+                try:
+                    has_archived = db.has_archived_messages(state.session_id)
+                except Exception:
+                    has_archived = False
+                db.replace_messages(
+                    state.session_id, state.history, active_only=has_archived
+                )
         except Exception:
             logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True)
 

hermes_state.py

@@ -3275,21 +3275,39 @@ class SessionDB:
             now_ts = max(now_ts + 1e-6, message_timestamp + 1e-6)
         return inserted, tool_calls_total
 
-    def replace_messages(self, session_id: str, messages: List[Dict[str, Any]]) -> None:
-        """Atomically replace every message for a session.
+    def replace_messages(
+        self,
+        session_id: str,
+        messages: List[Dict[str, Any]],
+        active_only: bool = False,
+    ) -> None:
+        """Atomically replace the stored messages for a session.
 
         Used by transcript-rewrite flows such as /retry, /undo, and /compress.
         The delete + reinsert sequence must commit as one transaction so a
         mid-rewrite failure does not leave SQLite with a partial transcript.
 
-        DESTRUCTIVE: the prior rows are DELETEd (and drop out of the FTS index).
-        For compaction that must preserve the pre-compaction transcript under
-        the same id, use :meth:`archive_and_compact` instead.
+        DESTRUCTIVE by default: every row for the session is DELETEd (and drops
+        out of the FTS index). For compaction that must preserve the
+        pre-compaction transcript under the same id, use
+        :meth:`archive_and_compact` instead.
+
+        Pass ``active_only=True`` to replace ONLY the live (``active = 1``) rows,
+        leaving soft-archived rows (``active = 0`` — e.g. the ``compacted = 1``
+        turns that :meth:`archive_and_compact` keeps on disk for #38763
+        durability, or rewind/undo rows) untouched. Callers that share a session
+        id with an agent already running in-place compaction must use this so a
+        full-history rewrite doesn't wipe the rows the agent deliberately
+        archived. ``message_count``/``tool_call_count`` then track the live set,
+        matching :meth:`archive_and_compact`.
         """
 
+        active_clause = " AND active = 1" if active_only else ""
+
         def _do(conn):
             conn.execute(
-                "DELETE FROM messages WHERE session_id = ?", (session_id,)
+                f"DELETE FROM messages WHERE session_id = ?{active_clause}",
+                (session_id,),
             )
             conn.execute(
                 "UPDATE sessions SET message_count = 0, tool_call_count = 0 WHERE id = ?",
@@ -3305,6 +3323,20 @@ class SessionDB:
 
         self._execute_write(_do)
 
+    def has_archived_messages(self, session_id: str) -> bool:
+        """Return True if the session has any soft-archived (``active = 0``) rows.
+
+        Used by callers (e.g. the ACP adapter's ``_persist``) that must decide
+        whether a full-history :meth:`replace_messages` would destroy durable
+        compaction-archived turns. Cheap existence probe — does not load rows.
+        """
+        with self._lock:
+            cursor = self._conn.execute(
+                "SELECT 1 FROM messages WHERE session_id = ? AND active = 0 LIMIT 1",
+                (session_id,),
+            )
+            return cursor.fetchone() is not None
+
     def archive_and_compact(
         self, session_id: str, compacted_messages: List[Dict[str, Any]]
     ) -> int:

tests/acp/test_session.py

@@ -330,6 +330,54 @@ class TestListAndCleanup:
             m["content"] for m in db.get_messages_as_conversation(state.session_id)
         ] == ["v2 replaced"]
 
+    def test_save_session_preserves_archived_rows_on_model_switch(self, tmp_path):
+        """Regression (#50405 W1/W2): a save by a fresh, non-self-persisting
+        agent must not destroy compaction-archived rows.
+
+        Model switches and /restore mint a brand-new agent with
+        ``_session_db_created=False`` (so it does NOT "own" persistence) and
+        then immediately call save_session. If the session had already
+        compacted, a blind full-history replace would DELETE the archived
+        active=0/compacted=1 rows — the same data loss the owned-agent guard
+        prevents. When archived rows exist, _persist must replace only the live
+        set (active_only) and leave the archived transcript intact.
+        """
+        from types import SimpleNamespace
+
+        db = SessionDB(tmp_path / "state.db")
+        # Use a mock agent factory so create_session doesn't spin up a real
+        # AIAgent (which needs credentials and leaks provider-probe state across
+        # xdist workers). The factory's agent does NOT own persistence to db.
+        manager = SessionManager(
+            agent_factory=lambda: SimpleNamespace(model="m"), db=db
+        )
+        state = manager.create_session(cwd="/work")
+
+        # Session flushed a live turn, then compaction archived it.
+        db.append_message(
+            session_id=state.session_id, role="user", content="archived needle"
+        )
+        db.archive_and_compact(
+            state.session_id, [{"role": "user", "content": "compacted summary"}]
+        )
+
+        # Model switch: a fresh agent bound to THIS db but not yet self-created.
+        state.agent = SimpleNamespace(
+            model="new-model", _session_db=db, _session_db_created=False
+        )
+        state.history = [{"role": "user", "content": "compacted summary"}]
+        manager.save_session(state.session_id)
+
+        # Archived pre-compaction turn survives and stays discoverable.
+        contents = [
+            m["content"]
+            for m in db.get_messages(state.session_id, include_inactive=True)
+        ]
+        assert "archived needle" in contents
+        assert "compacted summary" in contents
+        hits = {r["session_id"] for r in db.search_messages("needle")}
+        assert state.session_id in hits
+
     def test_cleanup_clears_all(self, manager):
         s1 = manager.create_session()
         s2 = manager.create_session()