fix(agent): observe commit phase without the fence lock (review F1)

begin_commit() retains the fence lock until finish_commit(), so a hung
SessionDB commit made try_cancel_before_commit() return None forever and
the host spun ahead of the overrun-warning loop — a genuinely hung commit
stayed unbounded AND silent. Add a lock-free phase marker (threading.Event
set inside begin_commit while the lock is held, readable without it) and
break the host spin on commit_in_flight so the bounded overrun loop — and
its WARNING + on_commit_overrun surfacing — is reachable WHILE the commit
is still blocked. Applies to both the sync compress wrapper and the
gateway session-hygiene wait.

Regression asserts the warning and callback fire while the event-gated
fake commit is still blocked; the test releases the worker only after
those assertions (addresses helix4u's released-before-asserting callout).

PR #76354 review, blocking finding 1 / merge gate 1.

agent/conversation_compression.py

@@ -432,6 +432,15 @@ class CompressionCommitFence:
         self._lock = threading.Lock()
         self._cancelled = False
         self._commit_started = False
+        # Lock-free commit-phase marker (#76354 review F1). ``begin_commit``
+        # RETAINS ``self._lock`` until ``finish_commit``, so any host-side
+        # observation that needs the lock (``try_cancel_before_commit``)
+        # blocks/space-outs for the whole commit. This Event is set inside
+        # ``begin_commit`` while the lock is held but is READABLE WITHOUT the
+        # lock, so a host can observe "a commit was admitted and may be in
+        # flight" even while the commit itself is hung — which is exactly when
+        # the overrun warning must be able to fire.
+        self._commit_phase = threading.Event()
         # Forward-progress telemetry: the compression worker touches this
         # whenever the streamed summary call produces a token (see
         # ContextCompressor._call_summary_llm). Waiters use it to distinguish
@@ -495,12 +504,28 @@ class CompressionCommitFence:
             self._lock.release()
             return False
         self._commit_started = True
+        # Set while the fence lock is held so observers can never see
+        # commit_in_flight=True for a commit that lost to cancellation.
+        self._commit_phase.set()
         return True
 
     def finish_commit(self) -> None:
         """Leave a commit boundary entered by :meth:`begin_commit`."""
+        self._commit_phase.clear()
         self._lock.release()
 
+    @property
+    def commit_in_flight(self) -> bool:
+        """Lock-free read: an admitted commit has begun and not yet finished.
+
+        Safe to call from the host while the worker holds the fence lock for
+        the whole commit (a hung SessionDB write). Hosts use this to reach
+        their overrun-warning loop WHILE the commit is blocked instead of
+        spinning on ``try_cancel_before_commit`` (which needs the lock the
+        worker retains until ``finish_commit``).
+        """
+        return self._commit_phase.is_set()
+
     @property
     def is_cancelled(self) -> bool:
         """True after cancellation won before the commit boundary."""
@@ -683,6 +708,14 @@ def run_compress_context_with_progress_timeout(
 
     cancelled: Optional[bool] = None
     while cancelled is None:
+        # F1: ``begin_commit`` retains the fence lock until ``finish_commit``,
+        # so a hung commit makes ``try_cancel_before_commit`` return None
+        # forever. The lock-free phase marker breaks the spin so the
+        # overrun-warning loop below is reachable WHILE the commit is still
+        # blocked.
+        if fence.commit_in_flight:
+            cancelled = False
+            break
         cancelled = fence.try_cancel_before_commit()
         if cancelled is None:
             time.sleep(0.001)

gateway/run.py

@@ -16671,6 +16671,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                     except asyncio.TimeoutError:
                                         _cancelled = None
                                         while _cancelled is None:
+                                            # #76354 F1: a hung commit retains the
+                                            # fence lock; the lock-free phase
+                                            # marker keeps this loop from spinning
+                                            # forever while the commit blocks.
+                                            if _hyg_commit_fence.commit_in_flight:
+                                                _cancelled = False
+                                                break
                                             _cancelled = (
                                                 _hyg_commit_fence.try_cancel_before_commit()
                                             )

tests/agent/test_compression_review_76354.py

@@ -0,0 +1,138 @@
+"""Regressions for the #76354 review of the compression timeout architecture.
+
+Every test here asserts the BLOCKED/hung state itself where the review demands
+it — the worker is released only AFTER the assertion (helix4u called out two
+prior tests that released before asserting; do not regress that).
+
+Covers:
+- F1: commit-phase overrun warning fires WHILE the commit is hung (lock-free
+  ``commit_in_flight`` phase marker).
+- F2: every host unwind (KeyboardInterrupt / generic exception) revokes commit
+  admission before the host resumes.
+- F4 (unit half): a cancelled attempt cannot clear the failure cooldown
+  (fence check ordered BEFORE cooldown-clear).
+- F6: bounded admission — four wedged workers refuse a fifth submission fast,
+  and the refused job never runs later; a cancelled fence skips summary work.
+- S3 analogue: the idle wait is charged from the last progress event, so
+  silence cannot approach 2x the configured idle timeout.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import logging
+import threading
+import time
+
+import pytest
+
+from agent.conversation_compression import (
+    CompressionCommitFence,
+    run_compress_context_with_progress_timeout,
+)
+
+
+def _drain_admission_slots():
+    """Placeholder until bounded admission (F6) lands in a later commit."""
+    return
+
+
+class TestF1CommitOverrunWhileHung:
+    def test_overrun_warning_fires_while_commit_still_blocked(self):
+        """The warning + on_commit_overrun fire DURING the hang, not after.
+
+        The fake commit is event-gated and is NOT released until after the
+        assertions on the callback/log have been made while the worker
+        thread is still blocked inside the commit boundary.
+        """
+        original = [{"role": "user", "content": "a"}]
+        compressed = [{"role": "assistant", "content": "late"}]
+        entered = threading.Event()
+        release = threading.Event()
+        overrun_fired = threading.Event()
+        overruns = []
+
+        def worker(fence: CompressionCommitFence):
+            assert fence.begin_commit()
+            entered.set()
+            try:
+                # Hung commit: blocked until the TEST releases it, which
+                # happens only after asserting the overrun surfaced.
+                assert release.wait(timeout=10)
+                return (compressed, "committed-late")
+            finally:
+                fence.finish_commit()
+
+        records = []
+
+        class _Capture(logging.Handler):
+            def emit(self, record):
+                records.append(record)
+
+        def on_overrun(waited, ceil):
+            overruns.append((waited, ceil))
+            overrun_fired.set()
+
+        done = {}
+
+        def run():
+            done["result"] = run_compress_context_with_progress_timeout(
+                worker=worker,
+                messages=original,
+                system_prompt_fallback="fallback",
+                idle_timeout_seconds=0.05,
+                total_ceiling_seconds=0.05,
+                on_commit_overrun=on_overrun,
+            )
+
+        comp_logger = logging.getLogger("agent.conversation_compression")
+        handler = _Capture(level=logging.WARNING)
+        comp_logger.addHandler(handler)
+        try:
+            t = threading.Thread(target=run, name="f1-hung-commit-host")
+            t.start()
+            try:
+                assert entered.wait(timeout=2)
+                # ── Assert WHILE the commit worker is still blocked ──────
+                assert overrun_fired.wait(timeout=5), (
+                    "on_commit_overrun must fire while the commit is hung"
+                )
+                assert not release.is_set()  # worker provably still blocked
+                assert t.is_alive()
+                deadline = time.time() + 5
+                while time.time() < deadline:
+                    if any(
+                        r.levelno >= logging.WARNING
+                        and "past the total ceiling" in r.getMessage()
+                        for r in list(records)
+                    ):
+                        break
+                    time.sleep(0.01)
+                overrun_logs = [
+                    r
+                    for r in list(records)
+                    if r.levelno >= logging.WARNING
+                    and "past the total ceiling" in r.getMessage()
+                ]
+                assert overrun_logs, (
+                    "expected the overrun WARNING while the commit was "
+                    f"still blocked; got: {[r.getMessage() for r in records]}"
+                )
+                assert overruns and overruns[0][1] == pytest.approx(0.05)
+            finally:
+                release.set()
+            t.join(timeout=5)
+            assert not t.is_alive()
+        finally:
+            comp_logger.removeHandler(handler)
+        assert done["result"] == (compressed, "committed-late")
+        _drain_admission_slots()
+
+    def test_commit_in_flight_marker_is_lock_free(self):
+        fence = CompressionCommitFence()
+        assert fence.commit_in_flight is False
+        assert fence.begin_commit()
+        # The fence lock is HELD here; the marker must still be readable.
+        assert fence.commit_in_flight is True
+        fence.finish_commit()
+        assert fence.commit_in_flight is False