feat(delegation): 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
(the design: one results block per fan-out). Failure is the case that
should not wait. In the 1,393-agent refactor run every wave-1 child died in
the 08:29 401 storm; 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 single largest idle gap of the run.

_run_children_parallel, for DETACHED batches only (the sync path prints a
completion line the parent is already watching), pushes ONE
type="async_delegation" event with task_failure_notice=True and a
single-entry results list when a child ends in a failure status while
siblings are still pending. Same event shape and routing fields as the
batch result, so every drain/ownership/format path treats it identically;
the batch record is not finalized and its consolidated result still
arrives unchanged. The formatter renders
"[ASYNC DELEGATION TASK FAILED — <id>, task i/n]" with the task, status,
error and live transcript path, and says the batch result is still coming.
The TUI dedup key distinguishes a notice from the final result and from a
sibling's notice.

Live through the real detached dispatch (3 children, one fails at 0.1 s,
two succeed at 8 s): notice drained at t+0.3 s, batch final at t+8.4 s. On
main the parent hears nothing until t+8.4 s.

Tests (2): the notice reaches the queue with the record's routing fields
while the record stays running, and formats as an early warning; no notice
for a finished batch; dedup key differs from the final result.
Delegation/registry/notification suites (129 files) 1,344 passed.

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)

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

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/<id>/.

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

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