Merge pull request #103549 from NousResearch/fix/async-batch-task-failure-notice

feat(delegation): a failed child of a running detached fan-out is surfaced immediately, as a non-durable notice (66 min dead wave; review-found final-result loss closed)

evals/postmortem/live_ab/batch_failure_notice.py

@@ -27,7 +27,9 @@ dd._Batch.run_child = fake_run_child
 # Background dispatch requires an async-capable session; emulate a CLI session key.
 import tools.async_delegation as ad
 t0 = time.time()
-out = dt.delegate_task(tasks=[{"goal": "fail fast on a simulated 401 error"}, {"goal": "slow successful worker number one"}, {"goal": "slow successful worker number two"}], background=True, parent_agent=parent)
+out = dt.delegate_task(# Grouped: siblings share ONE final result, so a dead sibling would otherwise wait for the slowest. (Ungrouped tasks are
+# their own async unit since the per-group split landed on main and already report as they finish.)
+tasks=[{"goal": "fail fast on a simulated 401 error", "group": "g"}, {"goal": "slow successful worker number one", "group": "g"}, {"goal": "slow successful worker number two", "group": "g"}], background=True, parent_agent=parent)
 print("dispatch:", json.dumps(json.loads(out))[:160])
 seen = []
 deadline = time.time() + 30

evals/postmortem/review_probes/notice_delivery_probe.py

@@ -35,9 +35,11 @@ class Sink(GatewayNotificationsMixin):
         return True
 
 def batch_run(did, gates):
-    tasks=[{'goal':f'worker task {i}'} for i in range(3)]
+    tasks=[{'goal':f'worker task {i}','group':'g'} for i in range(3)]  # grouped: one shared final, so the early notice matters
     children=[(i,t,SimpleNamespace()) for i,t in enumerate(tasks)]
     b=dd._Batch(tasks,children,SimpleNamespace(quiet_mode=True),{'model':'offline'},None,'leaf',3,did,[],[], '', '',None,None,time.monotonic())
+    # Since the per-group split on main, a detached unit carries its registry id; the notice keys on it.
+    if hasattr(b,'unit_id'): b.unit_id=did; b.group='g'
     def child(i,t,c):
         assert gates[i].wait(15)
         return {'task_index':i,'status':'error' if i<2 else 'completed','error':'offline failure' if i<2 else None,'summary':None if i<2 else 'FINISHED_SUCCESS','duration_seconds':0.1}

gateway/run_notifications.py

@@ -950,7 +950,14 @@ class GatewayNotificationsMixin:
         evt_type = str(evt.get("type") or "")
         if evt_type == "async_delegation":
             producer_id = str(evt.get("delegation_id") or "")
-            return (evt_type, producer_id, "") if producer_id else None
+            if not producer_id:
+                return None
+            if evt.get("task_failure_notice"):
+                # An interim per-task notice is its own producer event: it must not mark the
+                # batch's final result as already delivered, nor a sibling's notice.
+                task_idx = ((evt.get("results") or [{}])[0] or {}).get("task_index", "")
+                return (evt_type, producer_id, f"task_failure:{task_idx}")
+            return (evt_type, producer_id, "")
         if evt_type == "completion":
             producer_id = str(evt.get("session_id") or "")
             started_at = evt.get("started_at")
@@ -1035,7 +1042,9 @@ class GatewayNotificationsMixin:
         """
         claim = self._CompletionClaim()
         evt_type = evt.get("type")
-        if evt_type == "async_delegation":
+        # An interim per-task notice shares the batch's delegation_id but is not the durable
+        # completion; claiming that row here would acknowledge the FINAL result before it exists.
+        if evt_type == "async_delegation" and not evt.get("task_failure_notice"):
             claim.delegation_id = str(evt.get("delegation_id") or "")
             if claim.delegation_id:
                 try:

tests/tools/test_async_batch_task_failure_notice.py

@@ -0,0 +1,69 @@
+"""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)
+
+
+def test_interim_notice_never_claims_or_acknowledges_the_batch_final_row(tmp_path, monkeypatch):
+    """Independent-review witness: a busy parent that drained the notice first acknowledged the FINAL
+    result's durable row, and the consolidated result was never delivered (nor replayed after restart)."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hh"))
+    notice = {"type": "async_delegation", "delegation_id": "deleg_x", "task_failure_notice": True, "results": [{"task_index": 0}]}
+    final = {"type": "async_delegation", "delegation_id": "deleg_x", "is_batch": True, "results": []}
+    # The notice is a non-durable event: empty token, no row touched.
+    assert ad.claim_event_delivery(notice, "tui-poller") == ""
+    ad.complete_event_delivery(notice, "")
+    assert ad.is_interim_delegation_event(notice) and not ad.is_interim_delegation_event(final)
+
+
+def test_gateway_dedup_identity_separates_notices_from_the_final_and_from_each_other():
+    from gateway.run_notifications import GatewayNotificationsMixin
+    ident = GatewayNotificationsMixin._completion_delivery_identity
+    n0 = {"type": "async_delegation", "delegation_id": "d", "task_failure_notice": True, "results": [{"task_index": 0}]}
+    n1 = {"type": "async_delegation", "delegation_id": "d", "task_failure_notice": True, "results": [{"task_index": 1}]}
+    final = {"type": "async_delegation", "delegation_id": "d", "is_batch": True, "results": []}
+    assert len({ident(n0), ident(n1), ident(final)}) == 3

tools/async_delegation.py

@@ -361,8 +361,17 @@ def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool:
         return cur.rowcount == 1
 
 
+def is_interim_delegation_event(evt: Dict[str, Any]) -> bool:
+    """An early per-task notice for a batch that is still running. It shares the batch's
+    ``delegation_id`` but is NOT the durable completion: it must never claim, acknowledge or
+    dedup against the final result's row (independent review reproduced exactly that loss)."""
+    return evt.get("type") == "async_delegation" and bool(evt.get("task_failure_notice"))
+
+
 def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]:
-    """Claim a durable delegation event; non-durable events need no token."""
+    """Claim a durable delegation event; non-durable events (and interim notices) need no token."""
+    if is_interim_delegation_event(evt):
+        return ""
     delegation_id = str(evt.get("delegation_id") or "") if evt.get("type") == "async_delegation" else ""
     if not delegation_id:
         return ""
@@ -729,6 +738,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

@@ -135,6 +135,16 @@ def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interru
                     # Detached unit: a crash before the join must not lose children that already finished.
                     record_unit_child(batch.unit_id, entry)
                 _report_child_done(parent_agent, spinner_ref, entry, _tag, task_labels, n_tasks, n_here - len(results))
+                if (not honor_parent_interrupt and batch.unit_id and entry.get("status") in SUBAGENT_FAILURE_STATUSES
+                        and len(results) < n_here):
+                    # Detached unit, a sibling is still running: tell the parent NOW, not when the last one finishes.
+                    # Non-durable and separate from the unit's final result (which is still delivered once).
+                    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.unit_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:

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 []
@@ -165,6 +184,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))