From 8dc5608a7818ee4aabeaafc10b79d9652df27cf1 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 14 Aug 2026 20:59:29 -0700 Subject: [PATCH] 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 | 30 ++- agent/conversation_loop.py | 3 + gateway/session.py | 14 +- hermes_state.py | 9 + run_agent.py | 67 ++++- tests/gateway/test_session.py | 77 +++++- .../test_compression_closed_adoption.py | 242 ++++++++++++++++++ 7 files changed, 428 insertions(+), 14 deletions(-) create mode 100644 tests/run_agent/test_compression_closed_adoption.py diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index aeffedb678..45ede85c77 100644 --- a/agent/conversation_compression.py +++ b/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 diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index a27fd23d6b..53524fc357 100644 --- a/agent/conversation_loop.py +++ b/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 diff --git a/gateway/session.py b/gateway/session.py index 0121518152..cd62db5048 100644 --- a/gateway/session.py +++ b/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) diff --git a/hermes_state.py b/hermes_state.py index 342aae5394..2b5a66de7c 100644 --- a/hermes_state.py +++ b/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 ( diff --git a/run_agent.py b/run_agent.py index 4d6fc106e0..bc344b2cf0 100644 --- a/run_agent.py +++ b/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 diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index c81aade3f3..2defd546b7 100644 --- a/tests/gateway/test_session.py +++ b/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() diff --git a/tests/run_agent/test_compression_closed_adoption.py b/tests/run_agent/test_compression_closed_adoption.py new file mode 100644 index 0000000000..7309588467 --- /dev/null +++ b/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"