fix(sessions): recover compression parents without continuations

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

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,
         *,

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()

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(