49ed79bd54
Keep the positive case (interrupt returns without joining the wedged worker) and the control (uninterrupted batches still join and sort); the queued-child cancellation case duplicates the same shutdown call and exceeds the salvage bar of two invariant tests per fix.
90 lines
3.5 KiB
Python
90 lines
3.5 KiB
Python
"""A wedged child must not hold the parent past an interrupt (#116435).
|
|
|
|
``_run_children_parallel`` polled futures with a timeout and fabricated an
|
|
'interrupted' entry for still-pending children, but the executor's ``with``
|
|
exit then joined every worker — a child stuck in an uninterruptible call
|
|
(hung socket read) hung the parent thread forever, defeating the fast path.
|
|
The interrupt path now shuts the pool down without waiting.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
from tools.delegate_tool_dispatch import _Batch, _run_children_parallel
|
|
|
|
|
|
def _batch(children, parent):
|
|
tasks = [{"goal": f"task {i}"} for i in range(len(children))]
|
|
return _Batch(
|
|
task_list=tasks, children=[(i, tasks[i], c) for i, c in enumerate(children)],
|
|
parent_agent=parent, creds={}, context=None, top_role="leaf",
|
|
max_children=len(children), live_deleg_id=None, live_writers=[], live_paths=[],
|
|
origin_wake_sid="", origin_ui_session_id="", origin_owner_transport=None,
|
|
origin_owner_session_record=None, origin_session_history_delivery=False,
|
|
overall_start=time.monotonic(),
|
|
)
|
|
|
|
|
|
def test_interrupt_does_not_join_wedged_child():
|
|
wedged_release = threading.Event()
|
|
wedged_started = threading.Event()
|
|
parent = SimpleNamespace(
|
|
_interrupt_requested=False, _delegate_spinner=None, quiet_mode=True,
|
|
)
|
|
|
|
def run_child(i, task, child):
|
|
if i == 0:
|
|
wedged_started.set()
|
|
wedged_release.wait() # parked until test teardown
|
|
return {"task_index": i, "status": "completed"}
|
|
return {"task_index": i, "status": "completed", "summary": "done",
|
|
"error": None, "api_calls": 1, "duration_seconds": 0}
|
|
|
|
children = [SimpleNamespace(_delegate_role="leaf") for _ in range(2)]
|
|
batch = _batch(children, parent)
|
|
batch.run_child = run_child
|
|
results = []
|
|
returned = threading.Event()
|
|
|
|
def drive():
|
|
_run_children_parallel(batch, results, honor_parent_interrupt=True)
|
|
returned.set()
|
|
|
|
worker = threading.Thread(target=drive, daemon=True)
|
|
worker.start()
|
|
try:
|
|
assert wedged_started.wait(5), "wedged child never started"
|
|
assert parent._interrupt_requested is False
|
|
parent._interrupt_requested = True
|
|
# The poll loop ticks every 0.5s; 3s is generous and still catches a join.
|
|
assert returned.wait(3), "interrupt path joined the wedged worker"
|
|
assert len(results) == 2
|
|
by_index = {e["task_index"]: e for e in results}
|
|
assert by_index[0]["status"] == "interrupted"
|
|
assert by_index[1]["status"] in ("completed", "interrupted")
|
|
finally:
|
|
wedged_release.set()
|
|
worker.join(timeout=5)
|
|
|
|
|
|
def test_normal_completion_still_joins_cleanly():
|
|
"""No interrupt: all children finish and results sort by task_index."""
|
|
parent = SimpleNamespace(
|
|
_interrupt_requested=False, _delegate_spinner=None, quiet_mode=True,
|
|
)
|
|
|
|
def run_child(i, task, child):
|
|
return {"task_index": i, "status": "completed", "summary": f"s{i}",
|
|
"error": None, "api_calls": 1, "duration_seconds": 0}
|
|
|
|
children = [SimpleNamespace(_delegate_role="leaf") for _ in range(3)]
|
|
batch = _batch(children, parent)
|
|
batch.run_child = run_child
|
|
results = []
|
|
_run_children_parallel(batch, results, honor_parent_interrupt=True)
|
|
assert [e["task_index"] for e in results] == [0, 1, 2]
|
|
assert all(e["status"] == "completed" for e in results)
|