From 6767c06d347e5aac97dd64b5dfd8a534fb92d895 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 5 Sep 2026 01:55:22 -0700 Subject: [PATCH 1/5] feat(delegation): a failed child of a still-running detached fan-out is surfaced to the parent immediately MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 — , 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. --- .../test_async_batch_task_failure_notice.py | 48 +++++++++++++++++++ tools/async_delegation.py | 35 ++++++++++++++ tools/delegate_tool_dispatch.py | 16 +++++-- tools/process_registry_notifications.py | 21 ++++++++ tui_gateway/session_notifications.py | 4 ++ 5 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 tests/tools/test_async_batch_task_failure_notice.py 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)) From 70214e9ab8eef74234e168f6d92c678f2c369ec4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 5 Sep 2026 06:12:37 -0700 Subject: [PATCH 2/5] fix(delegation): an interim task-failure notice never claims, acknowledges or dedups against the batch's final result Independent review of the first version reproduced a final-result loss with real dispatch, SQLite claims and the delivery consumers: the interim notice carried the batch's delegation_id and every consumer claims and acknowledges durable rows by that id alone. A busy parent that drained the notice first acknowledged the FINAL result's row; the consolidated result could then never be claimed, and would not replay after restart. The gateway's in-memory dedup keyed on (type, delegation_id) and suppressed the final (and every sibling notice) the same way. The notice is now recognised as a non-durable event at both boundaries: claim_event_delivery returns the empty token for it (is_interim_delegation_event), the gateway preflight does not claim the durable row for it, and the gateway dedup identity carries the task index so notices are distinct from the final and from each other (the TUI key already did). Reviewer's own probe re-run on this head: gateway receives ['notice', 'notice', 'final'] (was ['notice']); busy-parent case: the notice takes no claim token, the final's claim succeeds and its row stays pending until delivered (was: final claim None, row 'delivered' with an undelivered result inside). Tests (2 new): the notice is non-durable at the claim boundary; the gateway identity separates two notices and the final into three. --- gateway/run_notifications.py | 13 ++++++++++-- .../test_async_batch_task_failure_notice.py | 21 +++++++++++++++++++ tools/async_delegation.py | 11 +++++++++- 3 files changed, 42 insertions(+), 3 deletions(-) diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index fc356f608c..c5ce3cb4b2 100644 --- a/gateway/run_notifications.py +++ b/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: diff --git a/tests/tools/test_async_batch_task_failure_notice.py b/tests/tools/test_async_batch_task_failure_notice.py index e1b9cce896..427855cab3 100644 --- a/tests/tools/test_async_batch_task_failure_notice.py +++ b/tests/tools/test_async_batch_task_failure_notice.py @@ -46,3 +46,24 @@ def test_notice_is_not_sent_for_a_finished_batch_and_does_not_dedup_against_the_ 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 diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 6fba6071da..97a0df13d4 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -323,8 +323,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 "" From 14c3101256f11d3e27cd41251cf1e0f6689f879d Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 6 Sep 2026 12:09:49 -0700 Subject: [PATCH 3/5] merge main: notice rides the per-group unit (unit_id, n_here); drop the detached kwarg main removed --- tools/delegate_tool_dispatch.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tools/delegate_tool_dispatch.py b/tools/delegate_tool_dispatch.py index 5d41ba42b5..09f8a42da4 100644 --- a/tools/delegate_tool_dispatch.py +++ b/tools/delegate_tool_dispatch.py @@ -94,7 +94,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, detached: bool = False) -> None: +def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interrupt: bool) -> 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 @@ -157,7 +157,7 @@ def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True if len(batch.children) == 1: results.append(batch.run_child(*batch.children[0])) else: - _run_children_parallel(batch, results, honor_parent_interrupt=honor_parent_interrupt, detached=detached) + _run_children_parallel(batch, results, honor_parent_interrupt=honor_parent_interrupt) _finalize_child_results(results, batch.task_list, batch.children, batch.parent_agent) total_duration = round(time.monotonic() - batch.overall_start, 2) From 8f2167342f5296d1283c230fb8726a5390e2a371 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 6 Sep 2026 12:12:09 -0700 Subject: [PATCH 4/5] evals(postmortem): notice probes target a grouped unit (per-group split on main; ungrouped tasks already report individually) --- evals/postmortem/live_ab/batch_failure_notice.py | 4 +++- evals/postmortem/review_probes/notice_delivery_probe.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/evals/postmortem/live_ab/batch_failure_notice.py b/evals/postmortem/live_ab/batch_failure_notice.py index d713e8f3bf..1e17dc895a 100644 --- a/evals/postmortem/live_ab/batch_failure_notice.py +++ b/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 diff --git a/evals/postmortem/review_probes/notice_delivery_probe.py b/evals/postmortem/review_probes/notice_delivery_probe.py index 8869b13f96..70a6a8f547 100644 --- a/evals/postmortem/review_probes/notice_delivery_probe.py +++ b/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} From c6bf8fd95ca37126c0873a9049910821957768d6 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 6 Sep 2026 12:37:59 -0700 Subject: [PATCH 5/5] chore: retrigger CI (zero-job dispatch failure, auto-heal)