fix(agent,gateway): charge the idle wait from the last progress event (review S3)

Both progress-aware waits (sync compress wrapper and gateway session
hygiene) slept a FULL idle interval and only then compared progress, so
progress early in an interval let silence approach 2x the configured
idle timeout before the waiter noticed. Compute each wait slice as
idle_timeout - elapsed_since_last_progress instead.

Regression: a worker that reports progress early and then goes silent is
timed out in ~1x the idle budget, not ~2x.

PR #76354 review, 'idle timeout can allow nearly twice that silence'.

agent/conversation_compression.py

@@ -885,7 +885,14 @@ def run_compress_context_with_progress_timeout(
             remaining_ceiling = ceiling - waited
             if remaining_ceiling <= 0:
                 break
-            wait_slice = min(idle, remaining_ceiling)
+            # #76354 S3 analogue for this wait: charge the idle budget from
+            # the LAST PROGRESS event, not from the start of this wait slice.
+            # Waiting a full ``idle`` after progress that landed early in the
+            # previous slice would allow silence to approach 2x the budget.
+            since_progress = fence.seconds_since_progress()
+            wait_slice = min(
+                max(idle - since_progress, 0.005), remaining_ceiling
+            )
             try:
                 result = future.result(timeout=wait_slice)
                 handled_exit = True

gateway/run.py

@@ -16644,10 +16644,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                         # the turn forever.
                                         _hyg_wait_started = time.monotonic()
                                         while True:
+                                            # #76354 S3: charge the idle budget
+                                            # from the LAST PROGRESS event, not
+                                            # from the start of this wait slice —
+                                            # otherwise silence can approach 2x
+                                            # the configured timeout.
+                                            _slice = max(
+                                                _hyg_timeout_seconds
+                                                - _hyg_commit_fence.seconds_since_progress(),
+                                                0.005,
+                                            )
                                             try:
                                                 _compressed, _ = await asyncio.wait_for(
                                                     asyncio.shield(_hyg_future),
-                                                    timeout=_hyg_timeout_seconds,
+                                                    timeout=_slice,
                                                 )
                                                 break
                                             except asyncio.TimeoutError:

tests/agent/test_compression_review_76354.py

@@ -409,3 +409,38 @@ class TestF6ExecutorSaturation:
             assert returned is messages
             # The cancelled attempt must not leave the durable lock held.
             assert db.get_compression_lock_holder(session_id) is None
+
+
+class TestS3IdleChargedFromLastProgress:
+    def test_silence_cannot_approach_double_idle_timeout(self):
+        """Progress early in an interval must not extend silence to ~2x idle."""
+        _drain_admission_slots()
+        idle = 0.4
+        release = threading.Event()
+
+        def worker(fence: CompressionCommitFence):
+            time.sleep(0.05)
+            fence.touch_progress()  # early progress, then total silence
+            assert release.wait(timeout=10)
+            return ([], "late")
+
+        t0 = time.monotonic()
+        try:
+            msgs, prompt = run_compress_context_with_progress_timeout(
+                worker=worker,
+                messages=[{"role": "user", "content": "a"}],
+                system_prompt_fallback="fb",
+                idle_timeout_seconds=idle,
+                total_ceiling_seconds=5.0,
+            )
+        finally:
+            elapsed = time.monotonic() - t0
+            release.set()
+        assert prompt == "fb"
+        # Old behavior waited a full interval from the CHECK (~2x idle ≈
+        # 0.85s+). New behavior times out ~idle after the last progress
+        # (~0.45s). Allow generous slack while still excluding ~2x.
+        assert elapsed < idle * 1.8, (
+            f"silence exceeded ~2x idle budget shape: {elapsed:.2f}s"
+        )
+        _drain_admission_slots()