fix(delegation): a crash mid-unit keeps the children that already finished

Since #104299 a background delegate_task call is split into completion units
(one per `group`, one per ungrouped task). A multi-child unit still joined on
all its children before anything was written durably, so an owner crash
between the first and last child lost the finished work and replayed the whole
unit as "outcome unknown" — the restart-granularity gap that #104233 (Xipong's
#76228/#76229 direction) solved with a second row per child.

Each finished child of a detached unit is now recorded on the unit's OWN row
(`record_unit_child` → result_json {results, partial}) as its future lands;
the real result overwrites it at finalize. `recover_abandoned_delegations`
replays recorded children with their real summaries and marks only the
unfinished ones unknown, naming the count. No new rows, no new consumer shape.
`task_indexes` is persisted so recovery knows a split unit's members.

tests/tools/test_async_delegation.py

@@ -1030,3 +1030,45 @@ def test_units_of_one_call_share_a_single_capacity_slot():
     assert (first["status"], second["status"], other["status"]) == ("dispatched", "dispatched", "rejected")
     assert ad.active_task_count() == 2
     gate.set()
+
+
+def test_child_finished_before_crash_is_recovered_with_its_result(tmp_path):
+    """Real-import E2E: a 2-task group unit whose owner dies mid-run replays the finished child's real result
+    and marks only the unfinished sibling unknown — a crash costs the stragglers, never the finished work."""
+    repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+    env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo}
+    producer = r'''
+import os, sys, time
+from unittest.mock import MagicMock
+import tools.delegate_tool as dt
+parent = MagicMock(); parent._delegate_depth = 0; parent.session_id = "sess"; parent._interrupt_requested = False
+parent._active_children = []; parent._active_children_lock = None
+def child(task_index, goal, child=None, parent_agent=None, **kw):
+    if task_index == 1:
+        time.sleep(600)
+    return {"task_index": task_index, "status": "completed", "summary": f"done: {goal}", "api_calls": 1,
+            "duration_seconds": 0.1, "model": "m", "exit_reason": "completed"}
+def build(**kw):
+    c = MagicMock(); c._delegate_role = "leaf"; c._subagent_id = f"s{kw['task_index']}"; return c
+creds = {"model": "m", "provider": None, "base_url": None, "api_key": None, "api_mode": None, "command": None, "args": None}
+dt._build_child_agent = build; dt._run_single_child = child; dt._resolve_delegation_credentials = lambda *a, **k: creds
+dt.delegate_task(tasks=[{"goal": "fast member of the group task", "group": "g"},
+                        {"goal": "slow member of the group task", "group": "g"}], background=True, parent_agent=parent)
+time.sleep(2.0)
+sys.stdout.flush(); os._exit(1)
+'''
+    subprocess.run([sys.executable, "-c", producer], cwd=repo, env=env, text=True, capture_output=True, timeout=30)
+    consumer = r'''
+import json, queue
+from tools import async_delegation as ad
+q = queue.Queue(); ad.restore_undelivered_completions(q)
+print(json.dumps(q.get_nowait(), sort_keys=True))
+'''
+    second = subprocess.run([sys.executable, "-c", consumer], cwd=repo, env=env, text=True, capture_output=True,
+                            timeout=15, check=True)
+    evt = json.loads(second.stdout.strip().splitlines()[-1])
+    by_index = {r["task_index"]: r for r in evt["results"]}
+    assert by_index[0]["status"] == "completed" and by_index[0]["summary"] == "done: fast member of the group task"
+    assert by_index[1]["status"] == "unknown"
+    assert "1/2 child results were recorded" in evt["error"]
+    assert "done: fast member" in format_process_notification(evt)

tools/async_delegation.py

@@ -168,7 +168,7 @@ def _persist_dispatch(record: Dict[str, Any]) -> None:
         owner_started_at = None
     task_payload = {
         key: record.get(key)
-        for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch", *_ROUTING_KEYS)
+        for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch", "task_indexes", *_ROUTING_KEYS)
         if key in record}
     with _DB_LOCK, _transaction() as conn:
         conn.execute("""INSERT OR REPLACE INTO async_delegations
@@ -218,8 +218,39 @@ def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None:
              json.dumps(event), json.dumps(result), event["delegation_id"]))
 
 
+def record_unit_child(delegation_id: str, entry: Dict[str, Any]) -> None:
+    """Durably record ONE finished child of a still-running multi-child unit on the unit's own row, so a crash before
+    the unit joins loses only the children that had not finished. Stored in ``result_json`` (overwritten by the real
+    result at finalize); ``recover_abandoned_delegations`` replays it. Best-effort: a failed write costs recovery
+    fidelity, never the live result."""
+    try:
+        with _DB_LOCK, _transaction() as conn:
+            row = conn.execute("SELECT result_json FROM async_delegations WHERE delegation_id=? AND state='running'",
+                               (delegation_id,)).fetchone()
+            if row is None:
+                return
+            partial = json.loads(row[0] or "{}") or {}
+            results = [r for r in partial.get("results") or [] if r.get("task_index") != entry.get("task_index")]
+            results.append(entry)
+            conn.execute("UPDATE async_delegations SET result_json=?, updated_at=? WHERE delegation_id=? AND state='running'",
+                         (json.dumps({"results": results, "partial": True}), time.time(), delegation_id))
+    except Exception:  # noqa: BLE001 — recovery bookkeeping must never fail a live child
+        logger.warning("Async delegation %s: could not record finished child %s", delegation_id, entry.get("task_index"), exc_info=True)
+
+
+def _recovered_results(task: Dict[str, Any], result_json: Optional[str], error: str) -> Optional[List[Dict[str, Any]]]:
+    """Per-task results for an abandoned unit: recorded children as they finished, the rest ``unknown``."""
+    partial = json.loads(result_json or "{}") or {}
+    if not (task.get("is_batch") and partial.get("partial") and partial.get("results")):
+        return None
+    recorded = {r["task_index"]: r for r in partial["results"] if isinstance(r.get("task_index"), int)}
+    indexes = task.get("task_indexes") or list(range(len(task.get("goals") or [])))
+    return [recorded.get(i) or {"task_index": i, "status": "unknown", "summary": None, "error": error} for i in indexes]
+
+
 def recover_abandoned_delegations() -> int:
-    """Classify records whose owning process disappeared as outcome unknown."""
+    """Classify records whose owning process disappeared as outcome unknown; children a multi-child unit had already
+    recorded (``record_unit_child``) are replayed with their real results."""
     try:
         from gateway.status import _pid_exists, get_process_start_time
     except Exception:
@@ -228,24 +259,31 @@ def recover_abandoned_delegations() -> int:
     with _DB_LOCK, _transaction() as conn:
         rows = conn.execute("""SELECT delegation_id, origin_session, origin_ui_session_id,
                       parent_session_id, dispatched_at, owner_pid,
-                      owner_started_at, task_json, origin_session_id
+                      owner_started_at, task_json, origin_session_id, result_json
                FROM async_delegations WHERE state IN ('running','finalizing')""").fetchall()
         for row in rows:
-            delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json, origin_sid = row
+            delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json, origin_sid, result_json = row
             if pid and _pid_exists(int(pid)) and (started is None or get_process_start_time(int(pid)) == int(started)):
                 continue
             task = json.loads(task_json or "{}")
+            error = "Delegation owner exited before recording a terminal result; outcome unknown."
+            recovered_results = _recovered_results(task, result_json, error)
+            if recovered_results:
+                done = sum(1 for r in recovered_results if r.get("status") != "unknown")
+                error = (f"Delegation owner exited before the unit finished; {done}/{len(recovered_results)} child "
+                         "results were recorded and are included below, the rest are unknown.")
             event = {
                 "type": "async_delegation", "delegation_id": delegation_id, "session_key": session_key,
                 "origin_ui_session_id": origin_ui, "origin_session_id": origin_sid or "",
                 "parent_session_id": parent_id, "goal": task.get("goal", ""), "goals": task.get("goals"),
                 "context": task.get("context"), "toolsets": task.get("toolsets"), "role": task.get("role"),
                 "model": task.get("model"), "is_batch": bool(task.get("is_batch")),
-                "status": "unknown", "summary": None,
-                "error": "Delegation owner exited before recording a terminal result; outcome unknown.",
+                "status": "unknown", "summary": None, "error": error,
+                **({"results": recovered_results} if recovered_results else {}),
                 "dispatched_at": dispatched_at, "completed_at": now,
                 **{k: task[k] for k in _ROUTING_KEYS if task.get(k)}}
-            result = {"status": "unknown", "summary": None, "error": event["error"]}
+            result = {"status": "unknown", "summary": None, "error": event["error"],
+                      **({"results": recovered_results} if recovered_results else {})}
             conn.execute("""UPDATE async_delegations SET state='unknown', completed_at=?,
                    updated_at=?, event_json=?, result_json=?, delivery_state='pending'
                    WHERE delegation_id=?""", (now, now, json.dumps(event), json.dumps(result), delegation_id))

tools/delegate_tool_dispatch.py

@@ -13,6 +13,7 @@ from concurrent.futures import FIRST_COMPLETED, wait as _cf_wait
 from dataclasses import dataclass, replace
 from typing import Any, Dict, List, Optional
 
+from tools.async_delegation import _new_delegation_id, record_unit_child
 from tools.delegate_tool_child_run import _detach_child, _fabricated_entry, _signal_child_stop
 from tools.delegate_tool_progress import (
     SUBAGENT_FAILURE_STATUSES, _clean_error_text, _print_completion_line, _quiet, format_batch_tag,
@@ -45,6 +46,7 @@ class _Batch:
     overall_start: float
     # Set on per-group units carved out by ``_dispatch_background``; None for the whole batch / ungrouped units.
     group: Optional[str] = None
+    unit_id: Optional[str] = None  # the async registry id this unit runs under (``<call_id>-k`` for split calls)
 
     def owner_kwargs(self) -> Dict[str, Any]:
         """Steer/stop authority of the originating session, passed to every child run."""
@@ -129,6 +131,9 @@ def _run_children_parallel(batch: _Batch, results: list, *, honor_parent_interru
             for future in done:
                 entry = _entry_of(future, futures[future])
                 results.append(entry)
+                if not honor_parent_interrupt and batch.unit_id:
+                    # 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))
     results.sort(key=lambda r: r["task_index"])  # match input order
 
@@ -363,6 +368,7 @@ def _dispatch_background(batch: _Batch) -> str:
         # One unit keeps the live-transcript directory's id so the returned delegation_id matches
         # cache/delegation/live/<id>/; several units suffix it (-1, -2, ...) and the call keeps the bare id.
         unit_id = batch.live_deleg_id if len(units) == 1 else (f"{batch.live_deleg_id}-{k + 1}" if batch.live_deleg_id else None)
+        unit.unit_id = unit_id = unit_id or _new_delegation_id()  # fixed before the runner can start
         dispatch = _dispatch_unit(unit, unit_id, slot_key, routing)
         if dispatch.get("status") == "dispatched":
             slot_key = slot_key or dispatch["delegation_id"]