From 0277cc48bd15d8d4d7b340f75ecf0947d9a9aaef Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 2 Aug 2026 15:26:16 -0700 Subject: [PATCH] fix(agent): never release the durable compression lease mid-commit (re-review #1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 56 ++++++++-- tests/agent/test_compression_review_76354.py | 108 +++++++++++++++++++ 2 files changed, 155 insertions(+), 9 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 756b93ecc49dd..4cce62313e969 100644 --- a/agent/conversation_compression.py +++ b/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 diff --git a/tests/agent/test_compression_review_76354.py b/tests/agent/test_compression_review_76354.py index cc06b428d60ab..abeb9153ccc9e 100644 --- a/tests/agent/test_compression_review_76354.py +++ b/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")