diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index eb7fa3a4c0..2d9b17fa80 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -1645,34 +1645,36 @@ def run_compress_context_with_progress_timeout( # so a NEW compressor can acquire the lock immediately (no ABA: the # DB release is holder-scoped). handled_exit = True - # #97488 teardown: give the cancelled worker a bounded grace to - # actually exit before this host moves on. The worker checks the - # poison fence between provider phases, so a cooperative worker - # exits quickly; an uninterruptible provider call is orphaned behind - # the fence after the grace elapses (its late result is discarded and - # cannot touch session state). - worker_exited = _join_cancelled_worker( - future, - min(_CANCELLED_WORKER_TEARDOWN_GRACE_SECONDS, ceiling), - ) - if worker_exited: - # The worker provably exited: no in-flight provider call can - # outlive this attempt, so the total-ceiling lease retention is - # no longer needed and a retry cannot overlap anything. - fence.allow_cancelled_lock_release() - else: - logger.warning( - "Cancelled compression worker did not exit within %.1fs " - "grace — orphaning it behind the poison fence (late result " - "will be discarded)%s", + # #97488 teardown (total-ceiling path only): give the cancelled + # worker a bounded grace to actually exit before this host moves on. + # The worker checks the poison fence between provider phases, so a + # cooperative worker exits quickly; an uninterruptible provider call + # is orphaned behind the fence after the grace elapses (its late + # result is discarded and cannot touch session state). The + # idle-stall path intentionally skips the join: its worker is by + # definition silent/hung, the stall-fallback retry below needs a + # prompt host return (pinned by the #76354 S3 latency contract), and + # the fence poison + attempt-generation supersession already protect + # state against its late unwind. + if total_exhausted: + worker_exited = _join_cancelled_worker( + future, min(_CANCELLED_WORKER_TEARDOWN_GRACE_SECONDS, ceiling), - ( - "; retaining the session compression lease until it " - "exits so no new attempt overlaps it" - if total_exhausted - else "" - ), ) + if worker_exited: + # The worker provably exited: no in-flight provider call can + # outlive this attempt, so the total-ceiling lease retention + # is no longer needed and a retry cannot overlap anything. + fence.allow_cancelled_lock_release() + else: + logger.warning( + "Cancelled compression worker did not exit within %.1fs " + "grace — orphaning it behind the poison fence (late " + "result will be discarded); retaining the session " + "compression lease until it exits so no new attempt " + "overlaps it", + min(_CANCELLED_WORKER_TEARDOWN_GRACE_SECONDS, ceiling), + ) fence.release_cancelled_compression_lock() waited = time.monotonic() - wait_started since_progress = fence.seconds_since_progress() diff --git a/tests/agent/test_compression_attempt_lifecycle.py b/tests/agent/test_compression_attempt_lifecycle.py index a1628da66d..d7cf68be4f 100644 --- a/tests/agent/test_compression_attempt_lifecycle.py +++ b/tests/agent/test_compression_attempt_lifecycle.py @@ -69,23 +69,29 @@ def _messages(): class TestWorkerTeardownOnCeiling: def test_cooperative_worker_joined_within_grace(self): - """A worker that exits promptly after cancel is joined; the lease is - released normally (no retention) — the sabotage check for this test - is removing the `_join_cancelled_worker` call, which makes - `worker_done_when_host_returned` False.""" + """A worker that exits promptly after cancel is joined on the + total-ceiling path; the lease is released normally (no retention) — + the sabotage check for this test is removing the + `_join_cancelled_worker` call, which makes + `worker_done.is_set()` False when the host returns.""" original = [{"role": "user", "content": "keep"}] worker_done = threading.Event() def cooperative_worker(fence: CompressionCommitFence): - # Poll the poison fence like the production worker does between - # provider phases; exit as soon as cancellation is visible. + # Continuous progress (the #97488 'last progress 0.0s ago' + # shape) so only the TOTAL ceiling expires; poll the poison + # fence like the production worker does between provider phases. deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: if fence.is_cancelled: break - fence_idle = fence.seconds_since_progress() - assert fence_idle >= 0 # precondition: fence clock live + fence.touch_progress() time.sleep(0.01) + # Cooperative-but-not-instant exit: the unwind after seeing the + # poison takes real time (rollback, telemetry). Long enough that + # a host WITHOUT the bounded-grace join returns first; far + # inside the 5s grace for a host WITH it. + time.sleep(0.08) worker_done.set() return (original, "late") @@ -94,8 +100,8 @@ class TestWorkerTeardownOnCeiling: worker=cooperative_worker, messages=original, system_prompt_fallback="fallback", - idle_timeout_seconds=0.05, - total_ceiling_seconds=0.15, + idle_timeout_seconds=0.1, + total_ceiling_seconds=0.2, fence=fence, stall_fallback=False, ) @@ -105,7 +111,11 @@ class TestWorkerTeardownOnCeiling: "host returned before tearing down a cooperative cancelled " "worker — bounded-grace join missing (#97488)" ) - assert prompt == "fallback" + # Whichever return path won the race (fallback via join, or the + # worker's own return adopted inside the final wait slice), the + # transcript must be unchanged. + assert msgs == [{"role": "user", "content": "keep"}] + assert prompt in ("fallback", "late") # Teardown proved quiescence, so the lease must NOT stay retained. assert fence._retain_cancelled_lock_until_worker_done is False