fix(agent): never release the durable compression lease mid-commit (re-review #1)

revoke_commit_admission() used to invoke the holder-qualified lease
release unconditionally — including while an admitted commit was still
mutating SessionDB — letting a second compressor acquire the durable
lock mid-commit and interleave with the first commit's writes.

The admission_revoked flag store stays lock-free, but the lease-release
decision now coordinates with the fence lock:
- revoke acquires the fence lock non-blocking; on success no commit can
  be in flight (an admitted commit retains the lock until finish_commit)
  and the release runs immediately, still under the lock so a racing
  begin_commit cannot slip between the check and the release.
- on failure the release is deferred: finish_commit() re-checks
  _admission_revoked and performs it AFTER the commit completes (prompt
  even if the worker thread is later parked), and the begin_commit
  refusal path does the same for a revoke that lost the race to a
  transient lock-setup/cancel boundary. All paths are idempotent with
  the worker's own outer cleanup (DB release is holder-qualified).

Invariant encoded + tested: no second compressor can acquire the durable
lock while an admitted commit is still mutating; after a post-revoke
commit finishes the lease is released promptly. Both regressions
(revoke-during-commit deferral, revoke-before-commit immediate release +
refused begin_commit) are sabotage-verified.

agent/conversation_compression.py

@@ -543,6 +543,12 @@ class CompressionCommitFence:
         ):
             self._cancelled = True
             self._lock.release()
+            if self._admission_revoked:
+                # Round-2 #1: a revoke that lost the fence-lock race to this
+                # very begin_commit deferred its lease release; the commit was
+                # refused, so the release is safe (and idempotent with the
+                # worker's own holder-qualified cleanup) right now.
+                self.release_cancelled_compression_lock()
             return False
         self._commit_started = True
         # Set while the fence lock is held so observers can never see
@@ -554,6 +560,15 @@ class CompressionCommitFence:
         """Leave a commit boundary entered by :meth:`begin_commit`."""
         self._commit_phase.clear()
         self._lock.release()
+        if self._admission_revoked:
+            # Round-2 #1: a revoke that arrived while THIS commit was in
+            # flight deferred its durable-lease release rather than freeing
+            # the lock out from under an active SessionDB mutation. The
+            # commit is now fully complete, so perform the deferred release
+            # here — promptly, without relying on the (possibly parked)
+            # worker thread's outer cleanup. Idempotent with that cleanup:
+            # the DB release is holder-qualified.
+            self.release_cancelled_compression_lock()
 
     @property
     def commit_in_flight(self) -> bool:
@@ -573,21 +588,44 @@ class CompressionCommitFence:
         return self._cancelled or self._admission_revoked
 
     def revoke_commit_admission(self) -> None:
-        """Revoke FUTURE commit admission without acquiring the fence lock.
+        """Revoke FUTURE commit admission without blocking on the fence lock.
 
         #76354 review F2: every host unwind path (KeyboardInterrupt, task
         cancellation, unexpected exception while waiting) must guarantee a
         detached worker cannot later enter the commit boundary and mutate
-        durable/session state. This is deliberately lock-free: a commit that
-        is ALREADY in flight cannot be safely abandoned (the invariant
-        "commit never abandoned mid-mutation" holds), but no NEW commit will
-        be admitted after this call — ``begin_commit`` re-checks the flag
-        under the fence lock. Also releases the worker's durable compression
-        lease via the holder-qualified hook when one was published (F4), so
-        a hung worker cannot retain the durable lock past a host unwind.
+        durable/session state. The flag store is lock-free: a commit that is
+        ALREADY in flight cannot be safely abandoned (the invariant "commit
+        never abandoned mid-mutation" holds), but no NEW commit will be
+        admitted after this call — ``begin_commit`` re-checks the flag under
+        the fence lock.
+
+        Round-2 #1 (durable-lease timing): the worker's holder-qualified
+        lease release (F4) must NOT run while an admitted commit is still
+        mutating SessionDB — a second compressor could otherwise acquire the
+        durable lock mid-commit and interleave with the first commit's
+        writes. The release decision is therefore made under the fence lock:
+
+        - non-blocking acquire succeeds → no commit is in flight (an
+          admitted commit RETAINS the lock until ``finish_commit``), so the
+          lease is released immediately, while still holding the lock so a
+          concurrent ``begin_commit`` cannot slip in between the check and
+          the release (it would be refused anyway — the flag is already set).
+        - acquire fails → the lock holder is either an in-flight commit or a
+          transient boundary (lock-setup / cancel admission). Defer: the
+          release then runs in ``finish_commit`` (after the mutation fully
+          completes) or on the ``begin_commit``-refusal path, whichever the
+          worker reaches first. Both are idempotent with the worker's own
+          outer cleanup because the DB release is holder-qualified.
         """
         self._admission_revoked = True
-        self.release_cancelled_compression_lock()
+        if self._lock.acquire(blocking=False):
+            try:
+                self.release_cancelled_compression_lock()
+            finally:
+                self._lock.release()
+        # else: deferred — finish_commit()/begin_commit() re-check
+        # _admission_revoked and perform the release once no commit can be
+        # mid-mutation.
 
     # ── Holder-qualified durable-lease cancellation (#76354 F4) ──────────
     # Transplanted from PR #71569 (@ciabata-git): the worker publishes an

tests/agent/test_compression_review_76354.py

@@ -444,3 +444,111 @@ class TestS3IdleChargedFromLastProgress:
             f"silence exceeded ~2x idle budget shape: {elapsed:.2f}s"
         )
         _drain_admission_slots()
+
+
+class TestRound2MidCommitLeaseRelease:
+    """Round-2 #1: revoke must not release the durable lease mid-commit.
+
+    Invariant: at no point can a second compressor acquire the durable lock
+    while an admitted commit is still mutating; after the commit finishes
+    post-revoke, the lease IS released promptly even if the worker thread is
+    later parked (never runs its outer cleanup).
+    """
+
+    def _db_with_lease(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        session_id = "R2_MID_COMMIT_LEASE"
+        db.create_session(session_id, source="cli")
+        holder = "pid:worker:original"
+        assert db.try_acquire_compression_lock(
+            session_id, holder, ttl_seconds=60
+        )
+        return db, session_id, holder
+
+    def test_revoke_during_in_flight_commit_defers_lease_release(
+        self, tmp_path
+    ):
+        """Event-gated fake commit; assertions run WHILE it is blocked."""
+        db, session_id, holder = self._db_with_lease(tmp_path)
+        fence = CompressionCommitFence()
+        fence.register_cancelled_lock_release(
+            lambda: db.release_compression_lock(session_id, holder)
+        )
+
+        commit_entered = threading.Event()
+        release_commit = threading.Event()
+        commit_finished = threading.Event()
+
+        def _committing_worker():
+            assert fence.begin_commit()
+            commit_entered.set()
+            assert release_commit.wait(timeout=10)
+            fence.finish_commit()
+            commit_finished.set()
+            # Park forever: the deferred release must NOT depend on this
+            # thread's outer cleanup running.
+            threading.Event().wait(30)
+
+        worker = threading.Thread(target=_committing_worker, daemon=True)
+        worker.start()
+        assert commit_entered.wait(timeout=5)
+
+        # Host revokes WHILE the commit is in flight.
+        fence.revoke_commit_admission()
+
+        # ── Assert the hung state BEFORE releasing the commit ────────────
+        assert not commit_finished.is_set()
+        assert db.get_compression_lock_holder(session_id) == holder, (
+            "revoke released the durable lease while a commit was still "
+            "mutating SessionDB"
+        )
+        assert not db.try_acquire_compression_lock(
+            session_id, "pid:second:contender", ttl_seconds=60
+        ), (
+            "a second compressor acquired the durable lock DURING an "
+            "admitted commit"
+        )
+
+        # ── Release the commit; deferred release must fire promptly ──────
+        release_commit.set()
+        assert commit_finished.wait(timeout=5)
+        deadline = time.time() + 5
+        while time.time() < deadline:
+            if db.get_compression_lock_holder(session_id) is None:
+                break
+            time.sleep(0.01)
+        assert db.get_compression_lock_holder(session_id) is None, (
+            "lease was not released promptly after the post-revoke commit "
+            "finished (worker thread is parked, so finish_commit must have "
+            "performed the deferred release)"
+        )
+        assert db.try_acquire_compression_lock(
+            session_id, "pid:second:contender", ttl_seconds=60
+        )
+        db.release_compression_lock(session_id, "pid:second:contender")
+
+    def test_revoke_before_commit_releases_immediately_and_refuses_commit(
+        self, tmp_path
+    ):
+        """No commit in flight → immediate release; begin_commit refused."""
+        db, session_id, holder = self._db_with_lease(tmp_path)
+        fence = CompressionCommitFence()
+        fence.register_cancelled_lock_release(
+            lambda: db.release_compression_lock(session_id, holder)
+        )
+
+        fence.revoke_commit_admission()
+
+        # Release happened synchronously inside revoke — no worker involved.
+        assert db.get_compression_lock_holder(session_id) is None, (
+            "revoke before begin_commit must release the lease immediately"
+        )
+        assert fence.begin_commit() is False, (
+            "begin_commit must be refused after admission was revoked"
+        )
+        assert db.try_acquire_compression_lock(
+            session_id, "pid:second:contender", ttl_seconds=60
+        )
+        db.release_compression_lock(session_id, "pid:second:contender")