fix(compression): adopt live continuation tip at flush across multi-hop chains

A turn writing against a session already closed by compression died with
session_persistence_failed and a misleading "this is often a full disk"
dialog, even though the store was healthy and a live continuation existed
(#82001). Depth-1 recovery (find_live_compression_child) could not resolve
lineages with >=2 compression hops (root -> mid -> tip), reproduced
independently on two- and three-hop chains.

- run_agent.py flush chokepoint: on CompressionSessionClosedError, resolve
  tip = db.get_compression_tip(old_id) (canonical bounded transitive walk),
  adopt only when tip != old_id AND the tip row is live, retry the flush
  exactly once (adoption budget); otherwise fail closed.
- gateway/session.py append_to_transcript: replace the depth-1 live-child
  lookup with the same tip + liveness contract, so gateway transcript
  reroutes follow full chains.
- agent/conversation_compression.py _adopt_live_compression_child: turn-start
  recovery preflight now resolves via get_compression_tip with the same
  liveness check, closing the last depth-1 consumer in this family.
- classify_persistence_error: new "compression_closed" bucket; the turn-end
  explanation names compression rotation and tells the client to refresh the
  session id instead of blaming a full disk.

Tests: depth-1 adoption, multi-hop chain adoption (agent + gateway), fail
closed with no continuation / stale-closed (ws_orphan_reap) tip, exactly-once
adoption budget, and error-wording guards (compression-closed never mentions
disk; real disk failures keep disk guidance).

Closes #82001

Co-authored-by: Al3xand3r1987 <125030427+Al3xand3r1987@users.noreply.github.com>
Co-authored-by: yuzilongleif-collab <235949691+yuzilongleif-collab@users.noreply.github.com>

agent/conversation_compression.py

@@ -1209,27 +1209,39 @@ def _adopt_live_compression_child(
     session_db: Any,
     parent_session_id: str,
 ) -> Optional[List[Dict[str, Any]]]:
-    """Move a stale compression contender onto the unique durable child.
+    """Move a stale compression contender onto the live continuation tip.
 
     Resolve and load first, then mutate the live agent. This ordering keeps the
     stale contender fail-closed when lineage is ambiguous or the compacted
     handoff cannot be read.
+
+    Resolution uses the canonical transitive walk ``get_compression_tip`` so a
+    lineage with >=2 compression hops (root -> mid -> tip) recovers to the live
+    tip — the depth-1 ``find_live_compression_child`` lookup this used to call
+    finds no live *direct* child in that shape and skipped recovery (#82001).
+    The tip walk returns the input id when no continuation exists, and a
+    resolved tip is adopted only while its row is still live — both cases fail
+    closed exactly as before.
     """
-    finder = getattr(type(session_db), "find_live_compression_child", None)
+    resolver = getattr(type(session_db), "get_compression_tip", None)
+    row_getter = getattr(type(session_db), "get_session", None)
     loader = getattr(type(session_db), "get_messages_as_conversation", None)
-    if not callable(finder) or not callable(loader):
+    if not callable(resolver) or not callable(row_getter) or not callable(loader):
         return None
-    child = finder(session_db, parent_session_id)
-    if not child or not child.get("id"):
+    tip = resolver(session_db, parent_session_id)
+    if not tip or str(tip) == str(parent_session_id):
+        return None
+    child_session_id = str(tip)
+    child = row_getter(session_db, child_session_id)
+    if not isinstance(child, dict) or child.get("ended_at") is not None:
         return None
-    child_session_id = str(child["id"])
     recovered = loader(session_db, child_session_id)
     if not isinstance(recovered, list) or not recovered:
         return None
-    # Revalidate after loading: the child may have rotated or a competing
+    # Revalidate after loading: the tip may have rotated or a competing
     # continuation may have appeared between the two DB reads.
-    confirmed = finder(session_db, parent_session_id)
-    if not confirmed or str(confirmed.get("id") or "") != child_session_id:
+    confirmed = resolver(session_db, parent_session_id)
+    if not confirmed or str(confirmed) != child_session_id:
         return None
 
     agent.session_id = child_session_id

agent/conversation_loop.py

@@ -1652,6 +1652,9 @@ def run_conversation(
     # Reset alongside the failure flag so a lock-contention diagnosis from a
     # previous turn can never leak into this turn's user-facing explanation.
     agent._last_persistence_error_cause = None
+    # Per-turn diagnostic: a failed compression-tip adoption in a previous
+    # turn's flush must not be reported against this turn.
+    agent._compression_adoption_failed = False
 
     # Main conversation loop counters (pure locals consumed by the loop below).
     api_call_count = 0

gateway/session.py

@@ -3567,8 +3567,18 @@ class SessionStore:
                 from hermes_state import CompressionSessionClosedError
 
                 if isinstance(exc, CompressionSessionClosedError):
-                    child = self._db.find_live_compression_child(session_id)
-                    child_id = str(child["id"]) if child and child.get("id") else ""
+                    # Resolve the full continuation chain via the canonical
+                    # transitive API — a depth-1 live-child lookup misses
+                    # lineages with >=2 compression hops (root -> mid -> tip).
+                    # ``get_compression_tip`` returns the input id when no
+                    # continuation exists; adopt only a different, still-live
+                    # tip, otherwise fail closed as before.
+                    child_id = ""
+                    tip = self._db.get_compression_tip(session_id)
+                    if tip and tip != session_id:
+                        tip_row = self._db.get_session(tip)
+                        if tip_row is not None and tip_row.get("ended_at") is None:
+                            child_id = str(tip)
                     if child_id:
                         try:
                             self._append_transcript_message(child_id, msg)

hermes_state.py

@@ -1495,6 +1495,7 @@ def is_disk_full_error(exc: BaseException | str | None) -> bool:
 PERSISTENCE_ERROR_CAUSES = (
     "locked",
     "compression",
+    "compression_closed",
     "turn_lease",
     "disk",
     "unknown",
@@ -1515,6 +1516,10 @@ def classify_persistence_error(exc_or_str) -> str:
       database write lock); transient, retry-later guidance applies.
     * ``"compression"`` — a live compression lease refused the transcript
       write; the database itself is healthy and unlocked.
+    * ``"compression_closed"`` — the write targeted a session already
+      rotated (closed) by compression and no live continuation was adopted;
+      the store is healthy — the client must refresh/adopt the new session
+      id, so disk-space advice would be a misdiagnosis.
     * ``"turn_lease"`` — a presented session-turn-lease holder no longer
       owns the conversation (expired, released, or reclaimed); fail-fast
       fencing, not a storage fault.
@@ -1532,11 +1537,15 @@ def classify_persistence_error(exc_or_str) -> str:
     # survived RPC wrapping).
     if isinstance(exc_or_str, SessionTurnLeaseLostError):
         return "turn_lease"
+    if isinstance(exc_or_str, CompressionSessionClosedError):
+        return "compression_closed"
     if isinstance(exc_or_str, CompressionSessionBusyError):
         return "compression"
     text = str(exc_or_str).lower()
     if "turn lease" in text:
         return "turn_lease"
+    if "closed by compression" in text:
+        return "compression_closed"
     if "being compressed" in text or "compression lease" in text:
         return "compression"
     if (

run_agent.py

@@ -2013,6 +2013,7 @@ class AIAgent:
         self,
         messages: List[Dict],
         conversation_history: Optional[List[Dict]] = None,
+        _adoption_budget: int = 1,
     ):
         """Persist any un-flushed messages to the SQLite session store.
 
@@ -2317,9 +2318,64 @@ class AIAgent:
             # before it is swallowed into a bare ``False`` — classify it here
             # so the turn-end explanation can distinguish lock contention
             # ("storage was busy, send it again") from disk-full/read-only.
-            from hermes_state import classify_persistence_error
+            from hermes_state import (
+                CompressionSessionClosedError,
+                classify_persistence_error,
+            )
 
             self._last_persistence_error_cause = classify_persistence_error(e)
+            if isinstance(e, CompressionSessionClosedError):
+                # Compression race: another path rotated this session while
+                # this turn was still writing against it. The store resolves
+                # the continuation chain transitively via the canonical API
+                # ``get_compression_tip`` (bounded walk, excludes branch/
+                # delegate/tool children, prefers live children over stale
+                # closed siblings such as ``ws_orphan_reap``). Adopt the tip
+                # ONLY when it is a different row AND still live, and retry
+                # the flush exactly once (adoption budget) — a second
+                # closed-parent write must fail closed, never loop. The tip
+                # walk returns the input id when no continuation exists, so
+                # ``tip == session_id`` means fail closed.
+                if _adoption_budget > 0:
+                    old_id = self.session_id
+                    tip = None
+                    try:
+                        tip = self._session_db.get_compression_tip(old_id)
+                    except Exception as tip_exc:
+                        logger.warning(
+                            "compression tip lookup failed for %s: %s",
+                            old_id,
+                            tip_exc,
+                        )
+                    if tip and tip != old_id:
+                        tip_row = None
+                        try:
+                            tip_row = self._session_db.get_session(tip)
+                        except Exception:
+                            tip_row = None
+                        if tip_row is not None and tip_row.get("ended_at") is None:
+                            logger.warning(
+                                "Adopted live compression tip %s for closed "
+                                "session %s; retrying flush once",
+                                tip,
+                                old_id,
+                            )
+                            self.session_id = tip
+                            self._flushed_db_message_ids = set()
+                            self._last_flushed_db_idx = 0
+                            self._compression_adoption_failed = False
+                            return self._flush_messages_to_session_db_unlocked(
+                                messages,
+                                conversation_history,
+                                _adoption_budget=0,
+                            )
+                # No live tip (or budget exhausted): fail closed — never guess
+                # a target session. The per-turn diagnostic flag lets the
+                # turn-completion explanation name compression rotation
+                # instead of the historical (misleading) full-disk advice.
+                self._compression_adoption_failed = True
+                logger.warning("Session DB append_message failed: %s", e)
+                return False
             logger.warning("Session DB append_message failed: %s", e)
             return False
 
@@ -3736,6 +3792,15 @@ class AIAgent:
                     "compressing this session. Your message should already be "
                     "saved — please send it again after compression completes."
                 )
+            if cause == "compression_closed":
+                return (
+                    prefix
+                    + "the turn was stopped because this session was rotated "
+                    "by context compression and its live continuation could "
+                    "not be adopted. The storage itself is healthy — refresh "
+                    "the client (or start a new turn) so it picks up the new "
+                    "session id, then send your message again."
+                )
             if cause == "turn_lease":
                 return (
                     prefix

tests/gateway/test_session.py

@@ -1401,15 +1401,88 @@ class TestGatewaySessionDbRecovery:
         ]
         db.close()
 
+    def test_transcript_reroute_follows_multi_hop_compression_chain(self, tmp_path):
+        """A stale writer behind >=2 compression hops (root -> mid -> tip) must
+        reroute to the live tip via the transitive ``get_compression_tip`` walk
+        — the depth-1 live-child lookup found nothing here (#82001)."""
+        import threading
+        from types import SimpleNamespace
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("root", source="telegram")
+        db.end_session("root", "compression")
+        db.create_session("mid", source="telegram", parent_session_id="root")
+        db.end_session("mid", "compression")
+        db.create_session("tip", source="telegram", parent_session_id="mid")
+        db.replace_messages("tip", [{"role": "user", "content": "summary"}])
+
+        store = object.__new__(SessionStore)
+        store._db = db
+        store._lock = threading.RLock()
+        store._entries = {"route": SimpleNamespace(session_id="root")}
+        store._loaded = True
+        store._save = lambda: None
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = False
+
+        store.append_to_transcript(
+            "root", {"role": "assistant", "content": "routed to tip"}
+        )
+
+        assert store._entries["route"].session_id == "tip"
+        assert "root" not in store._dirty_transcripts
+        assert [m["content"] for m in db.get_messages_as_conversation("root")] == []
+        assert [m["content"] for m in db.get_messages_as_conversation("tip")] == [
+            "summary",
+            "routed to tip",
+        ]
+        db.close()
+
+    def test_transcript_reroute_fails_closed_on_stale_closed_tip(self, tmp_path):
+        """A chain ending in a closed sibling (``ws_orphan_reap``) has no live
+        tip — the reroute must fail closed, never adopt a closed session."""
+        import threading
+        from types import SimpleNamespace
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("root", source="telegram")
+        db.end_session("root", "compression")
+        db.create_session("stale", source="telegram", parent_session_id="root")
+        db.end_session("stale", "ws_orphan_reap")
+
+        store = object.__new__(SessionStore)
+        store._db = db
+        store._lock = threading.RLock()
+        store._entries = {"route": SimpleNamespace(session_id="root")}
+        store._loaded = True
+        store._save = lambda: None
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = False
+
+        store.append_to_transcript(
+            "root", {"role": "assistant", "content": "must not land"}
+        )
+
+        assert store._entries["route"].session_id == "root"
+        assert [m["content"] for m in db.get_messages_as_conversation("stale")] == []
+        db.close()
+
     def test_transcript_reroute_migrates_remaining_backlog_to_child(self):
         import threading
         from types import SimpleNamespace
         from hermes_state import CompressionSessionClosedError
 
         class FakeDb:
-            def find_live_compression_child(self, session_id):
+            def get_compression_tip(self, session_id):
                 assert session_id == "parent"
-                return {"id": "child"}
+                return "child"
+
+            def get_session(self, session_id):
+                return {"id": session_id, "ended_at": None}
 
         store = object.__new__(SessionStore)
         store._db = FakeDb()

tests/run_agent/test_compression_closed_adoption.py

@@ -0,0 +1,242 @@
+"""Compression race at the flush chokepoint: a turn writing against a session
+already closed by compression must adopt the LIVE continuation tip instead of
+dying with ``session_persistence_failed`` and a misleading "full disk" dialog.
+
+The store resolves the continuation chain transitively via the canonical API
+``SessionDB.get_compression_tip`` (bounded walk, excludes branch/delegate/tool
+children, prefers live children over stale closed siblings). This suite proves
+the agent flush path:
+
+* adopts a unique live child (depth-1 case),
+* follows a chain of >=2 compressions to the live head — THE regression the
+  depth-1 ``find_live_compression_child`` API missed (#82001),
+* fails closed when no continuation exists (no retry loop),
+* fails closed when the resolved tip is itself closed (``ws_orphan_reap``),
+* performs the tip lookup exactly once per flush (adoption budget), and
+* never renders the failure with the historical full-disk misdiagnosis.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _flush_agent(db, session_id):
+    """Bind the real flush methods onto a stand-in over a live SessionDB."""
+    agent = SimpleNamespace(
+        _session_db=db,
+        _session_db_created=True,
+        _persist_disabled=False,
+        session_id=session_id,
+        _session_persist_lock=None,
+        _flushed_db_message_ids=set(),
+        _flushed_db_message_session_id=None,
+        _last_flushed_db_idx=0,
+        _db_flush_scan_prefix=None,
+        _persist_user_message_idx=None,
+        _persist_user_message_override=None,
+        _persist_user_message_timestamp=None,
+        _pending_cli_user_message=None,
+        _active_session_turn_lease_holder=None,
+        _last_persistence_error_cause=None,
+        _compression_adoption_failed=False,
+    )
+    agent._ensure_db_session = lambda: None
+    agent._flush_messages_to_session_db = (
+        AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent)
+    )
+    agent._flush_messages_to_session_db_unlocked = (
+        AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent)
+    )
+    return agent
+
+
+def _build_compression_chain(db: SessionDB, chain: list[str]) -> tuple[str, str]:
+    """Create ``chain[0] -> ... -> chain[-1]`` where every session except the
+    last is compression-ended and the last is live. Returns (root, live_head).
+    """
+    for i, sid in enumerate(chain):
+        parent = chain[i - 1] if i > 0 else None
+        db.create_session(sid, source="tui", parent_session_id=parent)
+        if i < len(chain) - 1:
+            db.end_session(sid, "compression")
+    return chain[0], chain[-1]
+
+
+def test_flush_adopts_unique_live_continuation(tmp_path: Path) -> None:
+    db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        db.create_session("parent", source="tui")
+        db.append_message("parent", "user", "before split")
+        db.end_session("parent", "compression")
+        db.create_session("child", source="tui", parent_session_id="parent")
+
+        agent = _flush_agent(db, "parent")
+        messages = [{"role": "user", "content": "steered after compression"}]
+        result = agent._flush_messages_to_session_db(messages, [])
+
+        assert result is True, "flush must succeed after adopting the continuation"
+        assert agent.session_id == "child"
+        durable = db.get_messages_as_conversation("child")
+        assert any(
+            m.get("content") == "steered after compression" for m in durable
+        ), "the user message must land in the child session, not be lost"
+        # The compression-closed parent stays immutable.
+        parent_rows = db.get_messages_as_conversation("parent")
+        assert not any(
+            m.get("content") == "steered after compression" for m in parent_rows
+        )
+        assert agent._compression_adoption_failed is False
+    finally:
+        db.close()
+
+
+def test_flush_adopts_live_head_across_compression_chain(tmp_path: Path) -> None:
+    """A stale writer behind a chain of >=2 compressions adopts the live head.
+
+    This is the exact lineage from #82001 (`root(compressed) -> mid(compressed)
+    -> tip(live)`) that a depth-1 live-child lookup cannot resolve, because the
+    direct child is itself already compression-ended.
+    """
+    db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        root, head = _build_compression_chain(db, ["root", "mid", "tip"])
+
+        agent = _flush_agent(db, root)
+        messages = [{"role": "user", "content": "steered after double rotation"}]
+        result = agent._flush_messages_to_session_db(messages, [])
+
+        assert result is True, "flush must succeed by adopting the chain head"
+        assert agent.session_id == head, "agent must move to the live chain head"
+        durable = db.get_messages_as_conversation(head)
+        assert any(
+            m.get("content") == "steered after double rotation" for m in durable
+        ), "the user message must land in the chain head, not be lost"
+    finally:
+        db.close()
+
+
+def test_flush_fails_closed_when_no_continuation(tmp_path: Path) -> None:
+    db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        db.create_session("parent", source="tui")
+        db.append_message("parent", "user", "before split")
+        db.end_session("parent", "compression")
+
+        agent = _flush_agent(db, "parent")
+        messages = [{"role": "user", "content": "steered after compression"}]
+        result = agent._flush_messages_to_session_db(messages, [])
+
+        assert result is False, "no continuation -> fail closed (never guess)"
+        assert agent.session_id == "parent", "session id must not change"
+        assert agent._compression_adoption_failed is True
+        assert agent._last_persistence_error_cause == "compression_closed"
+    finally:
+        db.close()
+
+
+def test_flush_fails_closed_when_tip_is_stale_closed(tmp_path: Path) -> None:
+    """The canonical tip walk may land on a stale closed sibling (e.g.
+    ``ws_orphan_reap``) — a non-live tip must NOT be adopted; fail closed."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        db.create_session("parent", source="tui")
+        db.append_message("parent", "user", "before split")
+        db.end_session("parent", "compression")
+        db.create_session("stale", source="tui", parent_session_id="parent")
+        db.end_session("stale", "ws_orphan_reap")
+
+        agent = _flush_agent(db, "parent")
+        messages = [{"role": "user", "content": "steered after compression"}]
+        result = agent._flush_messages_to_session_db(messages, [])
+
+        assert result is False, "non-live tip must fail closed (never adopt stale)"
+        assert agent.session_id == "parent"
+        assert agent._compression_adoption_failed is True
+    finally:
+        db.close()
+
+
+def test_flush_adopts_exactly_once_no_retry_loop(tmp_path: Path, monkeypatch) -> None:
+    """Adoption budget: the tip lookup runs at most once per flush, and a
+    second closed-parent write after adoption fails closed instead of looping.
+    """
+    from hermes_state import CompressionSessionClosedError
+
+    db = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        _build_compression_chain(db, ["root", "tip"])
+
+        agent = _flush_agent(db, "root")
+
+        tip_calls = {"count": 0}
+        orig_tip = SessionDB.get_compression_tip
+
+        def _counting_tip(self, session_id):
+            tip_calls["count"] += 1
+            return orig_tip(self, session_id)
+
+        monkeypatch.setattr(SessionDB, "get_compression_tip", _counting_tip)
+
+        # Every batch write raises closed — including the post-adoption retry
+        # against the live tip (simulating the tip rotating again mid-flush).
+        def _always_closed(self, *, session_id, messages, **kwargs):
+            raise CompressionSessionClosedError(session_id)
+
+        monkeypatch.setattr(SessionDB, "append_messages_batch", _always_closed)
+
+        messages = [{"role": "user", "content": "steered after compression"}]
+        result = agent._flush_messages_to_session_db(messages, [])
+
+        assert result is False, "second closed-parent write must fail closed"
+        assert tip_calls["count"] == 1, "tip lookup must happen exactly once"
+        assert agent._compression_adoption_failed is True
+    finally:
+        db.close()
+
+
+# ---------------------------------------------------------------------------
+# Diagnostics: the failure must never read like a disk problem.
+# ---------------------------------------------------------------------------
+
+
+def test_compression_closed_error_classifies_as_compression_closed() -> None:
+    from hermes_state import (
+        PERSISTENCE_ERROR_CAUSES,
+        CompressionSessionClosedError,
+        classify_persistence_error,
+    )
+
+    cause = classify_persistence_error(CompressionSessionClosedError("session-abc"))
+    assert cause == "compression_closed"
+    assert cause in PERSISTENCE_ERROR_CAUSES
+    # String form (post-RPC wrapping) classifies identically.
+    assert (
+        classify_persistence_error(str(CompressionSessionClosedError("session-abc")))
+        == "compression_closed"
+    )
+
+
+def test_compression_closed_wording_never_mentions_disk() -> None:
+    from hermes_state import CompressionSessionClosedError, classify_persistence_error
+
+    text = AIAgent._format_turn_completion_explanation(
+        "session_persistence_failed",
+        persistence_cause=classify_persistence_error(
+            CompressionSessionClosedError("session-abc")
+        ),
+    )
+    assert text, "an abnormal persistence failure must produce an explanation"
+    assert "disk" not in text.lower(), "compression-race message must not blame disk"
+    assert "compression" in text.lower(), "message must name compression rotation"
+
+
+def test_disk_cause_keeps_disk_guidance() -> None:
+    text = AIAgent._format_turn_completion_explanation(
+        "session_persistence_failed", persistence_cause="disk"
+    )
+    assert "full disk" in text, "real disk failures must keep disk guidance"