fix(sessions): fence expired orphan recovery leases

hermes_state.py

@@ -3715,18 +3715,6 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             ):
                 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.
@@ -3745,6 +3733,29 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             if child is not None:
                 return False
 
+            # refresh_compression_lock() deliberately lets an owner revive its
+            # own expired row. Reclaim that row inside this write transaction
+            # before reopening: refresh-first makes the lease active and aborts
+            # recovery; recovery-first deletes the holder identity so a later
+            # refresh cannot resurrect it.
+            now = time.time()
+            lock_row = conn.execute(
+                "SELECT holder, expires_at FROM compression_locks "
+                "WHERE session_id = ?",
+                (session_id,),
+            ).fetchone()
+            if lock_row is not None:
+                expires_at = lock_row["expires_at"]
+                if expires_at is None or float(expires_at) >= now:
+                    return False
+                deleted = conn.execute(
+                    "DELETE FROM compression_locks "
+                    "WHERE session_id = ? AND holder = ? AND expires_at = ?",
+                    (session_id, lock_row["holder"], expires_at),
+                )
+                if deleted.rowcount != 1:
+                    return False
+
             updated = conn.execute(
                 "UPDATE sessions SET ended_at = NULL, end_reason = NULL "
                 "WHERE id = ? AND ended_at IS NOT NULL "

tests/state/test_compression_lineage_guard.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import time
+
 import pytest
 
 from hermes_state import SessionDB
@@ -102,6 +104,44 @@ def test_reopen_orphaned_compression_session_fails_closed_with_active_lease(
     assert db.get_session("leased-parent")["end_reason"] == "compression"
 
 
+def test_reopen_orphaned_compression_session_reclaims_expired_lease(
+    db: SessionDB,
+) -> None:
+    _compression_parent(db, "expired-lease-parent")
+    now = time.time()
+    db._conn.execute(
+        "INSERT INTO compression_locks "
+        "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)",
+        ("expired-lease-parent", "old-compressor", now - 60, now - 30),
+    )
+    db._conn.commit()
+
+    assert db.reopen_orphaned_compression_session("expired-lease-parent") is True
+    assert db.refresh_compression_lock(
+        "expired-lease-parent", "old-compressor"
+    ) is False
+    assert db.get_compression_lock_holder("expired-lease-parent") is None
+
+
+def test_reopen_orphaned_compression_session_loses_to_expired_lease_refresh(
+    db: SessionDB,
+) -> None:
+    _compression_parent(db, "refreshed-lease-parent")
+    now = time.time()
+    db._conn.execute(
+        "INSERT INTO compression_locks "
+        "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)",
+        ("refreshed-lease-parent", "live-compressor", now - 60, now - 30),
+    )
+    db._conn.commit()
+
+    assert db.refresh_compression_lock(
+        "refreshed-lease-parent", "live-compressor"
+    ) is True
+    assert db.reopen_orphaned_compression_session("refreshed-lease-parent") is False
+    assert db.get_session("refreshed-lease-parent")["end_reason"] == "compression"
+
+
 
 
 def test_find_live_compression_child_ignores_non_continuation_children(