From 988f2baaf8f8c2fe0ac5fe83a1adcb6177ff4efa Mon Sep 17 00:00:00 2001 From: izumi0uu Date: Thu, 6 Aug 2026 21:47:48 +0800 Subject: [PATCH] fix(sessions): recover compression parents without continuations --- agent/conversation_compression.py | 21 ++++++ hermes_state.py | 64 +++++++++++++++++++ .../agent/test_compression_orphan_recovery.py | 42 ++++++++++++ tests/state/test_compression_lineage_guard.py | 60 +++++++++++++++++ 4 files changed, 187 insertions(+) create mode 100644 tests/agent/test_compression_orphan_recovery.py diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 697258ebf7486..af266e5c5f244 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -1310,6 +1310,27 @@ def recover_rotated_compression_session( return recovered holder = holder_getter(session_id) if callable(holder_getter) else None if not holder or attempt == 20: + if not holder: + orphan_reopener = getattr( + type(session_db), + "reopen_orphaned_compression_session", + None, + ) + if callable(orphan_reopener): + try: + if orphan_reopener(session_db, session_id): + logger.warning( + "compression recovery: reopened orphaned " + "session=%s with no continuation", + session_id, + ) + except Exception as exc: + logger.debug( + "orphaned compression session reopen failed " + "for %s: %s", + session_id, + exc, + ) return None time.sleep(0.05) return None diff --git a/hermes_state.py b/hermes_state.py index ca92e2ab1b2dc..08d084e4d16bb 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -3691,6 +3691,70 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) ).fetchall() return self._session_row_dict(rows[0]) if len(rows) == 1 else None + def reopen_orphaned_compression_session(self, session_id: str) -> bool: + """Reopen a compression parent only when no continuation was published. + + Compression publication is atomic in current builds, but older builds + could leave a closed parent behind after an interrupted handoff. This + recovery is deliberately conservative: an active compression lease or + any canonical child means the lineage is still owned by another path, + so the caller must fail closed instead of reopening the parent. + """ + if not session_id: + return False + + def _do(conn): + parent = conn.execute( + "SELECT ended_at, end_reason FROM sessions WHERE id = ?", + (session_id,), + ).fetchone() + if ( + parent is None + or parent["ended_at"] is None + or parent["end_reason"] != "compression" + ): + return False + + # An expired row is harmless: a publisher must revalidate its lease + # before committing, while an active row indicates a handoff may + # still be in flight. + active_lock = conn.execute( + "SELECT 1 FROM compression_locks " + "WHERE session_id = ? " + "AND (expires_at IS NULL OR expires_at >= ?) LIMIT 1", + (session_id, time.time()), + ).fetchone() + if active_lock is not None: + return False + + # Treat any direct non-branch/non-delegate/non-tool child as a + # continuation, regardless of its current ended state. Reopening + # in that case could create a second live head for one lineage. + child = conn.execute( + """ + SELECT 1 + FROM sessions + WHERE parent_session_id = ? + AND json_extract(COALESCE(model_config, '{}'), '$._branched_from') IS NULL + AND json_extract(COALESCE(model_config, '{}'), '$._delegate_from') IS NULL + AND COALESCE(source, '') != 'tool' + LIMIT 1 + """, + (session_id,), + ).fetchone() + if child is not None: + return False + + updated = conn.execute( + "UPDATE sessions SET ended_at = NULL, end_reason = NULL " + "WHERE id = ? AND ended_at IS NOT NULL " + "AND end_reason = 'compression'", + (session_id,), + ) + return updated.rowcount == 1 + + return bool(self._execute_write(_do)) + def publish_compression_child( self, *, diff --git a/tests/agent/test_compression_orphan_recovery.py b/tests/agent/test_compression_orphan_recovery.py new file mode 100644 index 0000000000000..8770c6c7be64b --- /dev/null +++ b/tests/agent/test_compression_orphan_recovery.py @@ -0,0 +1,42 @@ +"""Recovery for legacy compression parents with no continuation child.""" + +from types import SimpleNamespace + +from agent.conversation_compression import recover_rotated_compression_session +from hermes_state import CompressionSessionClosedError, SessionDB + + +def test_recover_rotated_compression_session_reopens_legacy_orphan(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session("orphan", source="cli") + db.append_message("orphan", "user", "before compression") + db.end_session("orphan", "compression") + agent = SimpleNamespace(_session_db=db, session_id="orphan") + + assert recover_rotated_compression_session(agent) is None + db.append_message("orphan", "user", "after recovery") + finally: + db.close() + + +def test_recover_rotated_compression_session_keeps_parent_closed_with_child( + tmp_path, +): + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session("parent", source="cli") + db.append_message("parent", "user", "before compression") + db.end_session("parent", "compression") + db.create_session("child", source="cli", parent_session_id="parent") + agent = SimpleNamespace(_session_db=db, session_id="parent") + + assert recover_rotated_compression_session(agent) is None + try: + db.append_message("parent", "user", "must stay closed") + except CompressionSessionClosedError: + pass + else: + raise AssertionError("compression parent with child was reopened") + finally: + db.close() diff --git a/tests/state/test_compression_lineage_guard.py b/tests/state/test_compression_lineage_guard.py index 9e7535d0b0348..bc00758d8afea 100644 --- a/tests/state/test_compression_lineage_guard.py +++ b/tests/state/test_compression_lineage_guard.py @@ -42,6 +42,66 @@ def test_find_live_compression_child_fails_closed_when_ambiguous(db: SessionDB) assert db.find_live_compression_child("parent") is None +def test_reopen_orphaned_compression_session_reopens_parent_without_child( + db: SessionDB, +) -> None: + _compression_parent(db, "orphan") + + assert db.reopen_orphaned_compression_session("orphan") is True + assert db.get_session("orphan")["ended_at"] is None + assert db.get_session("orphan")["end_reason"] is None + + db.append_message("orphan", "user", "recovered turn") + assert [m["content"] for m in db.get_messages("orphan")] == [ + "before split", + "recovered turn", + ] + + +def test_reopen_orphaned_compression_session_fails_closed_with_child( + db: SessionDB, +) -> None: + _compression_parent(db, "parent-with-child") + db.create_session("child", source="webui", parent_session_id="parent-with-child") + + assert db.reopen_orphaned_compression_session("parent-with-child") is False + parent = db.get_session("parent-with-child") + assert parent["end_reason"] == "compression" + assert parent["ended_at"] is not None + + +def test_reopen_orphaned_compression_session_ignores_non_continuation_children( + db: SessionDB, +) -> None: + _compression_parent(db, "parent-with-non-continuation-children") + db.create_session( + "branch", + source="webui", + parent_session_id="parent-with-non-continuation-children", + model_config={"_branched_from": "parent-with-non-continuation-children"}, + ) + db.create_session( + "delegate", + source="tool", + parent_session_id="parent-with-non-continuation-children", + model_config={"_delegate_from": "parent-with-non-continuation-children"}, + ) + + assert db.reopen_orphaned_compression_session( + "parent-with-non-continuation-children" + ) is True + + +def test_reopen_orphaned_compression_session_fails_closed_with_active_lease( + db: SessionDB, +) -> None: + _compression_parent(db, "leased-parent") + assert db.try_acquire_compression_lock("leased-parent", "compressor") + + assert db.reopen_orphaned_compression_session("leased-parent") is False + assert db.get_session("leased-parent")["end_reason"] == "compression" + + def test_find_live_compression_child_ignores_non_continuation_children(