From 498dc1934d9d2e66a7767d5e787ae5994b71fae1 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 09:56:34 +0530 Subject: [PATCH 1/3] test(harness): prove the flush-window bound without timing the runner The test gave the run a 0.2-second flush window and asserted it ended in under 1.5 seconds. Both numbers measured the runner, not the bound: on a slow CI runner the terminal's own delivery missed the tiny window and the run exited 4. The window is now 2.0 seconds, and emit_skipped_receipts waits on an event that is never set, so the run can end only because the window's wait_for cuts it off. An outer 30-second guard turns a broken bound into a failure rather than a hang. The other assertions are kept. TH-9515 Signed-off-by: khushalsonawat --- tests/harness/test_hosted_entrypoint.py | 34 ++++++++++++++----------- 1 file changed, 19 insertions(+), 15 deletions(-) diff --git a/tests/harness/test_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index 7fa19b0a2..de0bd90b6 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -3578,9 +3578,11 @@ def test_post_terminal_wire_block_is_bounded_by_the_remaining_flush_window() -> # `emit_skipped_receipts`/`receipt()` push with `deadline=None` -- a degraded-but-alive # events channel could previously retry every skipped scenario's receipt for the full # `RetryPolicy` budget regardless of how much of the flush window was already spent, well past - # the point the gateway tears the sandbox down. A tiny window plus a slow - # `emit_skipped_receipts` reproduces the shape without a real multi-second stall dominating the - # suite: `asyncio.wait_for`'s own cancellation cuts the sleep short well before it would run out. + # the point the gateway tears the sandbox down. Here `emit_skipped_receipts` never finishes on + # its own, so the run can end only because the flush window's `wait_for` cuts it off: that is + # the bound under test, whatever the runner's speed. The window is wide enough for the + # terminal's own delivery on a slow runner, and the outer guard turns a broken bound into a + # failure rather than a hang. async def scenario() -> None: scenarios = [ FakeScenario("first", "platform-first", [FakeSubGoal("holds", True)]), @@ -3589,27 +3591,29 @@ async def scenario() -> None: harness = _build_harness( scenarios=scenarios, cancel_on_scenario="first", instances=1 ) - harness.deps.flush_window_seconds = 0.2 + harness.deps.flush_window_seconds = 2.0 original_emit = HostedScheduler.emit_skipped_receipts + never = asyncio.Event() - async def slow_emit(self: HostedScheduler, result: RunResult) -> None: - await asyncio.sleep(2.0) + async def stalled_emit(self: HostedScheduler, result: RunResult) -> None: + await never.wait() await original_emit(self, result) - HostedScheduler.emit_skipped_receipts = slow_emit # type: ignore[method-assign] - started = time.monotonic() + HostedScheduler.emit_skipped_receipts = stalled_emit # type: ignore[method-assign] try: - code = await he.run_job( - harness.job_path, harness.source, harness.output, deps=harness.deps - ) + try: + code = await asyncio.wait_for( + he.run_job( + harness.job_path, harness.source, harness.output, deps=harness.deps + ), + timeout=30.0, + ) + except asyncio.TimeoutError: + pytest.fail("the post-terminal wire block was not bounded by the flush window") finally: HostedScheduler.emit_skipped_receipts = original_emit - elapsed = time.monotonic() - started - assert elapsed < 1.5, ( - f"post-terminal wire block was not bounded: {elapsed:.2f}s" - ) assert ( code == he.EXIT_OK ) # the terminal was already delivered before the window ran out From 5e35bf80a6e7b9924de5547c2d6059157177a8ca Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Wed, 11 Nov 2026 10:14:29 +0530 Subject: [PATCH 2/3] test(harness): time the flush-window cut from inside the stalled call The change review (Lane Q, round 1) showed the 30 s guard was the only time check left, so a cut-off 10 s or 25 s late still passed. The stalled receipts call now records when it starts and when the window's cancel reaches it, re-raises the cancel, and the test asserts that gap is under twice the window. Both late plants now fail at that assertion. The never-reached receipt push and its assertion are dropped. TH-9515 Signed-off-by: khushalsonawat --- tests/harness/test_hosted_entrypoint.py | 27 ++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/tests/harness/test_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index de0bd90b6..e479bfd11 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -3579,10 +3579,12 @@ def test_post_terminal_wire_block_is_bounded_by_the_remaining_flush_window() -> # events channel could previously retry every skipped scenario's receipt for the full # `RetryPolicy` budget regardless of how much of the flush window was already spent, well past # the point the gateway tears the sandbox down. Here `emit_skipped_receipts` never finishes on - # its own, so the run can end only because the flush window's `wait_for` cuts it off: that is - # the bound under test, whatever the runner's speed. The window is wide enough for the - # terminal's own delivery on a slow runner, and the outer guard turns a broken bound into a - # failure rather than a hang. + # its own (the stall stands in for the receipt push), so the run can end only when the flush + # window's `wait_for` cancels it. The test times that cut from inside the stalled call, from its + # start to the cancel, so the check is set by the window's timer, not the runner's speed; the + # slack of twice the window still fails a cut that comes seconds late. The window is wide + # enough for the terminal's own delivery on a slow runner, and the outer guard turns a broken + # bound into a failure rather than a hang. async def scenario() -> None: scenarios = [ FakeScenario("first", "platform-first", [FakeSubGoal("holds", True)]), @@ -3595,10 +3597,15 @@ async def scenario() -> None: original_emit = HostedScheduler.emit_skipped_receipts never = asyncio.Event() + stall: dict[str, float] = {} async def stalled_emit(self: HostedScheduler, result: RunResult) -> None: - await never.wait() - await original_emit(self, result) + stall["entered"] = time.monotonic() + try: + await never.wait() + except asyncio.CancelledError: + stall["cut"] = time.monotonic() + raise # swallowing the cancel would break the bound under test HostedScheduler.emit_skipped_receipts = stalled_emit # type: ignore[method-assign] try: @@ -3625,9 +3632,11 @@ async def stalled_emit(self: HostedScheduler, result: RunResult) -> None: key[1]: body["status"] for key, body in harness.transport.receipts.items() } assert statuses.get("first") == "passed" - assert ( - "second" not in statuses - ) # window ran out before the skipped receipt could go + + assert "cut" in stall, "the stalled receipts were never cancelled" + window = harness.deps.flush_window_seconds + # Cut at the window, not seconds later. + assert stall["cut"] - stall["entered"] < 2 * window assert ( harness.provisioner.closed is True From 62b316b4321f49db636512abc6116cc2a9af67e2 Mon Sep 17 00:00:00 2001 From: khushalsonawat Date: Thu, 12 Nov 2026 12:10:32 +0530 Subject: [PATCH 3/3] test(harness): check the flush-window bound the code sets, not the time it took Dileep found the test still timed the cut, with enough slack that a cut 1.8 s late on a 2.0 s window passed, and that it couldn't tell the remaining window from the full one. The test now records the timeout passed to wait_for around the stalled emit_skipped_receipts and asserts 0 < timeout < window: the code passes the remaining window, which is always under the full one. The outer 30 s guard stays, so a hang still fails. The +1.8 plant and timeout=None both fail it. TH-9515 Signed-off-by: khushalsonawat --- tests/harness/test_hosted_entrypoint.py | 47 +++++++++++++++++-------- 1 file changed, 33 insertions(+), 14 deletions(-) diff --git a/tests/harness/test_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index e479bfd11..6fab887cb 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/tests/harness/test_hosted_entrypoint.py @@ -3580,11 +3580,13 @@ def test_post_terminal_wire_block_is_bounded_by_the_remaining_flush_window() -> # `RetryPolicy` budget regardless of how much of the flush window was already spent, well past # the point the gateway tears the sandbox down. Here `emit_skipped_receipts` never finishes on # its own (the stall stands in for the receipt push), so the run can end only when the flush - # window's `wait_for` cancels it. The test times that cut from inside the stalled call, from its - # start to the cancel, so the check is set by the window's timer, not the runner's speed; the - # slack of twice the window still fails a cut that comes seconds late. The window is wide - # enough for the terminal's own delivery on a slow runner, and the outer guard turns a broken - # bound into a failure rather than a hang. + # window's `wait_for` cancels it. The test checks the bound itself, not elapsed time: a spy on + # `wait_for` records the `timeout` the code passes with `emit_skipped_receipts`, and that must + # be the REMAINING window -- `deadline - now`, where the deadline is the arm time plus the + # window and the terminal's own flush has already run since arming. So the tightest true bound + # is `0 < timeout < window`: `None`, a padded timeout, or the full window (which would not + # shrink as the window is spent) all fail, whatever the runner's speed. The outer guard turns + # a missing bound into a failure rather than a hang. async def scenario() -> None: scenarios = [ FakeScenario("first", "platform-first", [FakeSubGoal("holds", True)]), @@ -3598,24 +3600,37 @@ async def scenario() -> None: original_emit = HostedScheduler.emit_skipped_receipts never = asyncio.Event() stall: dict[str, float] = {} + bounds: list[float | None] = [] + real_wait_for = asyncio.wait_for async def stalled_emit(self: HostedScheduler, result: RunResult) -> None: - stall["entered"] = time.monotonic() try: await never.wait() except asyncio.CancelledError: stall["cut"] = time.monotonic() raise # swallowing the cancel would break the bound under test + async def spying_wait_for(aw: Any, timeout: float | None = None) -> Any: + # `he.asyncio` is the shared asyncio module, so this also sees the outer guard below; + # record only the call that wraps the stalled receipts. + code = getattr(aw, "cr_code", None) + if code is stalled_emit.__code__: + bounds.append(timeout) + return await real_wait_for(aw, timeout=timeout) + HostedScheduler.emit_skipped_receipts = stalled_emit # type: ignore[method-assign] try: try: - code = await asyncio.wait_for( - he.run_job( - harness.job_path, harness.source, harness.output, deps=harness.deps - ), - timeout=30.0, - ) + with mock.patch.object(he.asyncio, "wait_for", spying_wait_for): + code = await real_wait_for( + he.run_job( + harness.job_path, + harness.source, + harness.output, + deps=harness.deps, + ), + timeout=30.0, + ) except asyncio.TimeoutError: pytest.fail("the post-terminal wire block was not bounded by the flush window") finally: @@ -3635,8 +3650,12 @@ async def stalled_emit(self: HostedScheduler, result: RunResult) -> None: assert "cut" in stall, "the stalled receipts were never cancelled" window = harness.deps.flush_window_seconds - # Cut at the window, not seconds later. - assert stall["cut"] - stall["entered"] < 2 * window + # Exactly one bounded wait, and its timeout is the remaining window: positive, and + # strictly under the full window because the terminal flush ran after the window armed. + assert len(bounds) == 1, bounds + bound = bounds[0] + assert bound is not None, "the receipts wait was given no timeout" + assert 0 < bound < window, (bound, window) assert ( harness.provisioner.closed is True