"""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)