diff --git a/tests/tools/test_async_batch_task_failure_notice.py b/tests/tools/test_async_batch_task_failure_notice.py new file mode 100644 index 0000000000..e1b9cce896 --- /dev/null +++ b/tests/tools/test_async_batch_task_failure_notice.py @@ -0,0 +1,48 @@ +"""A failed child of a still-running detached fan-out is surfaced to the parent immediately. + +Batches join on the slowest sibling before ONE consolidated block re-enters. In a 1,393-agent run every wave-1 +child died in a 401 storm at 08:29 and the parent learned of it at 09:36 from the batch's "unknown outcome" +block: 66 minutes of a dead wave with nothing running. +""" +import queue +from unittest.mock import patch + +from tools import async_delegation as ad +from tools.process_registry_notifications import format_process_notification +from tui_gateway.session_notifications import _notification_event_dedup_key + + +def _record(status="running"): + return {"delegation_id": "deleg_x", "status": status, "is_batch": True, "goals": ["a", "b", "c"], "goal": "a", + "session_key": "sk", "origin_ui_session_id": "ui", "origin_session_id": "", "parent_session_id": "root", + "dispatched_at": 1.0, "role": "leaf", "model": "m", "context": None, "toolsets": None} + + +def test_failure_notice_reaches_the_queue_while_the_batch_keeps_running_and_formats_as_early_warning(): + q = queue.Queue() + entry = {"task_index": 1, "status": "error", "error": "401 authentication_error: key invalid", "duration_seconds": 12.5, + "live_transcript": "/tmp/live/task-1.log"} + with patch.object(ad, "_records", {"deleg_x": _record()}), \ + patch("tools.process_registry.process_registry") as reg: + reg.completion_queue = q + ad.push_task_failure_notice("deleg_x", entry, n_tasks=3) + assert ad._records["deleg_x"]["status"] == "running" # not finalized + evt = q.get_nowait() + assert evt["type"] == "async_delegation" and evt["task_failure_notice"] is True + assert (evt["session_key"], evt["origin_ui_session_id"], evt["parent_session_id"]) == ("sk", "ui", "root") + text = format_process_notification(evt) + assert text.startswith("[ASYNC DELEGATION TASK FAILED — deleg_x, task 2/3]") + assert "Task: b" in text and "401 authentication_error" in text and "/tmp/live/task-1.log" in text + assert "consolidated results will still arrive" in text + + +def test_notice_is_not_sent_for_a_finished_batch_and_does_not_dedup_against_the_final_result(): + q = queue.Queue() + with patch.object(ad, "_records", {"deleg_x": _record(status="completed")}), \ + patch("tools.process_registry.process_registry") as reg: + reg.completion_queue = q + ad.push_task_failure_notice("deleg_x", {"task_index": 0, "status": "error"}, n_tasks=3) + assert q.empty() + notice = {"type": "async_delegation", "delegation_id": "deleg_x", "task_failure_notice": True, "results": [{"task_index": 2}]} + final = {"type": "async_delegation", "delegation_id": "deleg_x", "is_batch": True, "results": []} + assert _notification_event_dedup_key(notice) != _notification_event_dedup_key(final) diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 1ef7fbdc27..6fba6071da 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -678,6 +678,41 @@ def _push_completion_event(record: Dict[str, Any], result: Dict[str, Any], statu "result lost: %s", record.get("delegation_id"), exc) +def push_task_failure_notice(delegation_id: str, entry: Dict[str, Any], *, n_tasks: int) -> None: + """Surface ONE failed child of a still-running detached batch to the parent now, instead of + when the slowest sibling finishes. In a 1,393-agent run every wave-1 child died in a 401 storm + at 08:29 and the parent learned of it at 09:36, when the batch's "unknown outcome" block finally + arrived: 66 minutes of a dead wave with nothing running. The notice rides the same + ``type="async_delegation"`` event shape as the batch result (so every drain/route/format path + treats it identically) with ``task_failure_notice=True`` and a single-entry ``results`` list; the + batch record is NOT finalized and its consolidated result still arrives as before.""" + with _records_lock: + record = _records.get(delegation_id) + if record is None or record.get("status") not in _ACTIVE_STATES: + return + snapshot = dict(record) + try: + from tools.process_registry import process_registry + except Exception as exc: # pragma: no cover + logger.error("Async delegation batch %s: task failure notice dropped (process_registry import): %s", delegation_id, exc) + return + evt = { + "type": "async_delegation", "task_failure_notice": True, "is_batch": True, "n_tasks": n_tasks, + "delegation_id": delegation_id, "results": [entry], + "session_key": snapshot.get("session_key", ""), + "origin_ui_session_id": snapshot.get("origin_ui_session_id", ""), + "origin_session_id": snapshot.get("origin_session_id", ""), + "parent_session_id": snapshot.get("parent_session_id"), + "goal": snapshot.get("goal", ""), "goals": snapshot.get("goals"), "context": snapshot.get("context"), + "toolsets": snapshot.get("toolsets"), "role": snapshot.get("role"), "model": snapshot.get("model"), + "status": "running", "dispatched_at": snapshot.get("dispatched_at") or time.time(), "completed_at": time.time(), + **{k: snapshot[k] for k in _ROUTING_KEYS if snapshot.get(k)}} + try: + process_registry.completion_queue.put(evt) + except Exception as exc: # pragma: no cover + logger.error("Async delegation batch %s: failed to enqueue task failure notice: %s", delegation_id, exc) + + # ── Stale monitor ─────────────────────────────────────────────────────────── def _ensure_stale_monitor() -> None: """Start (once) the stale-delegation monitor thread. One daemon thread serves diff --git a/tools/delegate_tool_dispatch.py b/tools/delegate_tool_dispatch.py index 5f883a2b8f..e6f89170c2 100644 --- a/tools/delegate_tool_dispatch.py +++ b/tools/delegate_tool_dispatch.py @@ -89,7 +89,7 @@ def _report_child_done(parent_agent, spinner_ref, entry, tag, task_labels, n_tas with _quiet("Spinner update_text failed: %s"): spinner_ref.update_text(f"🔀 {'[' + tag + '] ' if tag else ''}{remaining} task{'s' if remaining != 1 else ''} remaining") -def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interrupt: bool) -> None: +def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interrupt: bool, detached: bool = False) -> None: """Run the batch's children in parallel, appending entries to ``results`` (sorted by task_index on return, one completion line printed per child). Polls futures with a short ``wait()`` timeout instead of ``as_completed()`` so a wedged child cannot block the parent forever after an interrupt; on parent interrupt the still-pending @@ -126,9 +126,17 @@ def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interru entry = _entry_of(future, futures[future]) results.append(entry) _report_child_done(parent_agent, spinner_ref, entry, _tag, task_labels, n_tasks, n_tasks - len(results)) + if detached and entry.get("status") in SUBAGENT_FAILURE_STATUSES and len(results) < n_tasks: + # The parent is not watching a spinner: tell it NOW, not when the last sibling finishes. + with _quiet("task failure notice failed", exc_info=True): + from tools.async_delegation import push_task_failure_notice + _i = entry.get("task_index", -1) + _live = batch.live_paths[_i] if isinstance(_i, int) and 0 <= _i < len(batch.live_paths) else None + push_task_failure_notice( + batch.live_deleg_id, {**entry, **({"live_transcript": _live} if _live else {})}, n_tasks=n_tasks) results.sort(key=lambda r: r["task_index"]) # match input order -def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True) -> dict: +def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True, detached: bool = False) -> dict: """Run all built children, join, finalize (hooks + cost rollup), return the combined dict. Shared by the sync path and the background runner: even in the background the batch JOINS on itself here so ONE consolidated results block re-enters the conversation. Live transcripts are finalized but retained as the full-fidelity record @@ -138,7 +146,7 @@ def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True if len(batch.task_list) == 1: results.append(batch.run_child(*batch.children[0])) else: - _run_children_parallel(batch, results, honor_parent_interrupt=honor_parent_interrupt) + _run_children_parallel(batch, results, honor_parent_interrupt=honor_parent_interrupt, detached=detached) _finalize_child_results(results, batch.task_list, batch.children, batch.parent_agent) total_duration = round(time.monotonic() - batch.overall_start, 2) @@ -319,7 +327,7 @@ def _dispatch_background(batch: _Batch) -> str: role=batch.top_role, model=batch.creds["model"], session_key=session_key, origin_ui_session_id=origin_ui_session_id, origin_session_id=wake_sid, parent_session_id=getattr(parent_agent, "session_id", None), - runner=lambda: _execute_and_aggregate(batch, honor_parent_interrupt=False), + runner=lambda: _execute_and_aggregate(batch, honor_parent_interrupt=False, detached=True), interrupt_fn=_batch_interrupt, max_async_children=_get_max_async_children(), # Reuse the live-transcript directory's id (when created) so the returned delegation_id matches # cache/delegation/live//. diff --git a/tools/process_registry_notifications.py b/tools/process_registry_notifications.py index 15c9355867..1821ea0199 100644 --- a/tools/process_registry_notifications.py +++ b/tools/process_registry_notifications.py @@ -113,6 +113,25 @@ def _preamble(evt: dict, title: str, intro: str, completed_at: float, *, with_go return lines +def _format_task_failure_notice(evt: dict, deleg_id: str) -> str: + """One child of a still-running fan-out failed: say which, why, and that the batch goes on.""" + (r,) = (evt.get("results") or [{}])[:1] or [{}] + goals, idx, n = evt.get("goals") or [], r.get("task_index", 0), evt.get("n_tasks") or len(evt.get("goals") or []) + goal = goals[idx] if idx < len(goals) else r.get("goal", "") + err = str(r.get("error") or "").strip().replace("\n", " ")[:400] + lines = [ + f"[ASYNC DELEGATION TASK FAILED — {deleg_id}, task {idx + 1}/{n}]", + "One subagent in a background fan-out you dispatched has failed while its siblings are still running. " + "The batch's consolidated results will still arrive when the last sibling finishes; this is an early " + "warning so you can re-dispatch or investigate now instead of then.", + f"Task: {goal}" if goal else "", + f"Status: {r.get('status', '?')} Duration: {r.get('duration_seconds', '?')}s" + (f"\nError: {err}" if err else ""), + ] + if r.get("live_transcript"): + lines.append(f"Live transcript: {r['live_transcript']}") + return "\n".join(line for line in lines if line) + + def _format_batch_delegation(evt: dict, deleg_id: str, completed_at: float) -> str: """Consolidated block for a delegate_task fan-out that finished as one unit.""" results, goals = evt.get("results") or [], evt.get("goals") or [] @@ -163,6 +182,8 @@ def _format_async_delegation(evt: dict) -> str: and result, so an agent deep in unrelated context can act on it or re-dispatch.""" deleg_id = evt.get("delegation_id", "unknown") completed_at = evt.get("completed_at") or time.time() + if evt.get("task_failure_notice"): + return _format_task_failure_notice(evt, deleg_id) if evt.get("is_batch") or isinstance(evt.get("results"), list): return _format_batch_delegation(evt, deleg_id, completed_at) status, summary, error = evt.get("status") or "completed", evt.get("summary"), evt.get("error") diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 3457bfa7ed..e7f6887d42 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -104,6 +104,10 @@ def _notification_event_dedup_key(evt: dict) -> tuple: evt_type = evt.get("type", "completion") if evt_type == "async_delegation": # No process session_id: else every completion keys as ("", "async_delegation") and the second is suppressed forever. + # An early per-task failure notice must not collapse with the batch's final result (nor with a sibling's notice). + if evt.get("task_failure_notice"): + task_idx = ((evt.get("results") or [{}])[0] or {}).get("task_index", "") + return (evt.get("delegation_id", ""), evt_type, "task_failure", task_idx) return (evt.get("delegation_id", ""), evt_type) extra = _DEDUP_EXTRA_FIELDS.get("watch_overflow_" if evt_type.startswith("watch_overflow_") else evt_type, ()) return (evt.get("session_id", ""), evt_type, *(evt.get(f, 0 if f == "suppressed" else "") for f in extra))