From f5c2ea49a4716ccd377c99511ee1bb9be1b56275 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 14 Jul 2026 23:36:37 -0700 Subject: [PATCH] test: deflake async-delegation interrupt test under CI load (#64767) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test: deflake async-delegation interrupt test under CI load test_interrupt_all_signals_running_children failed twice on a loaded CI worker: the blocker's ev.wait(timeout=5) expired before interrupt_all() ran, the record finalized on its own, and interrupt_all() found nothing running (n == 0, interrupted count 0 — the exact CI assertion failure, reproduced locally by inserting a 5.6s dispatch->interrupt gap). Raise the internal safety timeouts from 5s to 60s across the file's gated runners — they exist only as runaway guards (every test releases its gate explicitly); the pytest-level timeout is the real backstop. Same flake class as the removed test_crashed_runner_produces_error_ completion (#64431). * test: fix cross-test completion-event leak (second flake mechanism) The first CI failure was the 5s guard-timeout race; the rerun exposed a SECOND mechanism with a different signature: 'completed' == 'interrupted'. Prior tests (e.g. test_dispatch_rejected_at_capacity) release their gate and return immediately, but their workers finalize asynchronously — on a loaded runner the teardown drain races the in-flight _finalize, and the straggler 'completed' events leak into the NEXT test's queue, where _drain_one() picks one up instead of the interrupt event. Reproduced locally: gate release + immediate drain + 0.15s finalize delay leaked 2 events. Fixes: - teardown waits (bounded 2s) for active workers to finalize before draining, so events land in the owning test - the interrupt test matches its OWN delegation_id via _drain_for() instead of taking whatever event arrives first --- tests/tools/test_async_delegation.py | 57 ++++++++++++++++++++++------ 1 file changed, 46 insertions(+), 11 deletions(-) diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index cc4f7651d364..37aae286394a 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -25,6 +25,13 @@ def _clean_state(): while not process_registry.completion_queue.empty(): process_registry.completion_queue.get_nowait() yield + # Give just-released workers a beat to finalize BEFORE draining, so their + # completion events land now instead of leaking into the next test's + # queue (worker threads push events asynchronously; a drain that races an + # in-flight _finalize misses it). + deadline = time.monotonic() + 2.0 + while ad.active_count() and time.monotonic() < deadline: + time.sleep(0.02) ad._reset_for_tests() while not process_registry.completion_queue.empty(): process_registry.completion_queue.get_nowait() @@ -39,11 +46,30 @@ def _drain_one(timeout=5.0): return None +def _drain_for(delegation_id, timeout=5.0): + """Drain until the event for *delegation_id* appears (discarding others). + + Completion events are pushed asynchronously by worker threads, so a + straggler from a PREVIOUS test can land after that test's teardown drain + and leak into the current test's queue. Matching on delegation_id makes + the assertion immune to that cross-test leak. + """ + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not process_registry.completion_queue.empty(): + evt = process_registry.completion_queue.get_nowait() + if evt.get("delegation_id") == delegation_id: + return evt + continue + time.sleep(0.02) + return None + + def test_dispatch_returns_immediately_without_blocking(): gate = threading.Event() def runner(): - gate.wait(timeout=5) + gate.wait(timeout=60) return {"status": "completed", "summary": "done", "api_calls": 1, "duration_seconds": 0.1, "model": "m"} @@ -70,7 +96,7 @@ def test_async_executor_workers_are_daemon_threads(): gate = threading.Event() def runner(): - gate.wait(timeout=5) + gate.wait(timeout=60) return {"status": "completed", "summary": "done"} res = ad.dispatch_async_delegation( @@ -148,7 +174,7 @@ def test_dispatch_rejected_at_capacity(): ev = threading.Event() def blocker(): - ev.wait(timeout=5) + ev.wait(timeout=60) return {"status": "completed", "summary": "x"} for i in range(2): @@ -170,9 +196,15 @@ def test_dispatch_rejected_at_capacity(): def test_interrupt_all_signals_running_children(): ev = threading.Event() interrupted = {"count": 0} + # No short internal timeout: the blocker holds until interrupt_fn fires. + # The old ev.wait(timeout=5) made this test a change-detector for CI + # worker load — on a CPU-starved runner the 5s expired before + # interrupt_all() ran, the record finalized, and interrupt_all() found + # nothing running (n == 0). The pytest-level timeout is the real + # runaway guard. def blocker(): - ev.wait(timeout=5) + ev.wait(timeout=60) return {"status": "interrupted", "summary": None, "error": "cancelled"} @@ -180,7 +212,7 @@ def test_interrupt_all_signals_running_children(): interrupted["count"] += 1 ev.set() - ad.dispatch_async_delegation( + r = ad.dispatch_async_delegation( goal="long task", context=None, toolsets=None, role="leaf", model="m", session_key="", runner=blocker, interrupt_fn=interrupt_fn, max_async_children=3, @@ -188,8 +220,11 @@ def test_interrupt_all_signals_running_children(): n = ad.interrupt_all(reason="test") assert n == 1 assert interrupted["count"] == 1 - # child still emits a completion event after interrupt - evt = _drain_one() + # child still emits a completion event after interrupt. Match on THIS + # delegation's id — straggler 'completed' events from a previous test's + # workers can finalize after that test's teardown drain and leak into + # this queue (observed on loaded CI workers). + evt = _drain_for(r["delegation_id"]) assert evt is not None assert evt["status"] == "interrupted" @@ -408,7 +443,7 @@ def test_delegate_task_background_routes_async_and_does_not_block(monkeypatch): gate = threading.Event() def slow_child(task_index, goal, child=None, parent_agent=None, **kw): - gate.wait(timeout=5) # a sync impl would hang delegate_task here + gate.wait(timeout=60) # a sync impl would hang delegate_task here return { "task_index": 0, "status": "completed", "summary": f"done: {goal}", "api_calls": 1, "duration_seconds": 0.1, "model": "m", @@ -536,7 +571,7 @@ def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch): gate = threading.Event() def _blocking_child(task_index, goal, child=None, parent_agent=None, **kw): - gate.wait(timeout=5) + gate.wait(timeout=60) return { "task_index": task_index, "status": "completed", "summary": f"done: {goal}", "api_calls": 1, @@ -690,7 +725,7 @@ def test_delegate_task_background_detaches_child_from_parent(monkeypatch): gate = threading.Event() def slow_child(task_index, goal, child=None, parent_agent=None, **kw): - gate.wait(timeout=5) + gate.wait(timeout=60) return {"task_index": 0, "status": "completed", "summary": "ok"} def build_and_register(**kw): @@ -722,7 +757,7 @@ def test_concurrent_dispatch_respects_capacity(): gate = threading.Event() def blocker(): - gate.wait(timeout=5) + gate.wait(timeout=60) return {"status": "completed", "summary": "x"} results = []