fix(delegation): a one-child detached unit keeps its finished child on owner death

`recover_abandoned_delegations` only replays recorded children of a unit, and only
the >1-child join path ever called `record_unit_child` (c5594ec4b3 wired it into
`_run_children_parallel`). The single-child branch of `_execute_and_aggregate` ran
the child with no durable record at all, so a background `delegate_task` with one
goal — the default shape, and every ungrouped unit under `independent_completions` —
lost a finished child whenever the owner died between the child's return and the
durable completion write (OOM-kill on Linux, orphaning elsewhere, and everything
after the child returns is owner-lifetime: host-owned finalize, transcripts, manifest,
then the write). The parent got the bare "outcome unknown" and the work was gone (#116000).

`_record_finished_child` is now the one place that decides to durably record a finished
child of a detached unit; the join loop and the one-child branch both call it.
No new rows, no new consumer shape.

Tests: tests/tools/test_async_delegation.py::test_one_child_unit_keeps_its_finished_child_when_the_owner_dies

tests/tools/test_async_delegation.py

@@ -1128,6 +1128,65 @@ print(json.dumps(q.get_nowait(), sort_keys=True))
     assert "done: fast member" in format_process_notification(evt)
 
 
+def test_one_child_unit_keeps_its_finished_child_when_the_owner_dies(tmp_path):
+    """#116000: a detached unit with exactly ONE child had NO durable record of that child at all — only the
+    multi-child join path called ``record_unit_child`` — so an owner death (OOM-kill / orphaning) anywhere in the
+    window after the child returned (host-owned finalize, transcripts, manifest, then the durable completion write)
+    replayed a bare "outcome unknown" and threw the finished work away. Real-import E2E: a one-task background
+    ``delegate_task`` child completes, the owner is killed while blocked inside that window, and a fresh process
+    must replay the child's real result to the parent."""
+    repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+    marker = tmp_path / "child-returned.flag"
+    env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo, "REPRO_MARKER": str(marker)}
+    producer = r'''
+import os, sys, time
+from unittest.mock import MagicMock
+import tools.delegate_tool as dt
+import tools.delegate_tool_dispatch as dtd
+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):
+    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
+def held_finalize(*a, **k):
+    # The child's result exists; the owner still has host-owned finalize + transcripts + manifest + the durable
+    # write to do. Block HERE so the driver kills the owner inside that window: deterministic, no race.
+    open(os.environ["REPRO_MARKER"], "w").write("child-returned")
+    time.sleep(600)
+dtd._finalize_child_results = held_finalize
+dt.delegate_task(tasks=[{"goal": "single background subagent"}], background=True, parent_agent=parent)
+time.sleep(600)
+'''
+    proc = subprocess.Popen([sys.executable, "-u", "-c", producer], cwd=repo, env=env,
+                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    try:
+        deadline = time.monotonic() + 60
+        while not marker.exists() and proc.poll() is None and time.monotonic() < deadline:
+            time.sleep(0.05)
+        assert marker.exists(), "the child never returned, so the death window was never reached"
+    finally:
+        proc.kill()
+        proc.wait(timeout=20)
+        time.sleep(0.3)  # let the OS reap the owner before recovery asks whether its pid is alive
+    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, "-u", "-c", consumer], cwd=repo, env=env, text=True,
+                            capture_output=True, timeout=30, check=True)
+    evt = json.loads(second.stdout.strip().splitlines()[-1])
+    (entry,) = evt["results"]  # the finished child, not a fabricated "unknown"
+    assert entry["status"] == "completed" and entry["summary"] == "done: single background subagent"
+    assert "1/1 child results were recorded" in evt["error"]
+    assert "done: single background subagent" in format_process_notification(evt)
+
+
 @pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX mode bits not enforced on Windows")
 def test_connect_creates_state_db_0o600_under_permissive_umask(tmp_path, monkeypatch):
     """``_connect`` shares state.db with hermes_state.SessionDB -- a fresh

tools/delegate_tool_dispatch.py

@@ -106,6 +106,14 @@ 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 _record_finished_child(batch: _Batch, entry: Any, honor_parent_interrupt: bool) -> None:
+    """Detached (background) unit: durably record a child on the unit's own row the moment it finishes, so an owner
+    death before the unit's join — or anywhere before its durable completion write, the whole remaining window for a
+    one-child unit — loses only children still running, never finished work (#116000). Best-effort by construction:
+    ``record_unit_child`` never raises into the join."""
+    if not honor_parent_interrupt and batch.unit_id and isinstance(entry, dict):
+        record_unit_child(batch.unit_id, entry)
+
 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()``
@@ -150,9 +158,8 @@ 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)
+                # Detached unit: a crash before the join must not lose children that already finished.
+                _record_finished_child(batch, entry, honor_parent_interrupt)
                 _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):
@@ -179,6 +186,9 @@ def _execute_and_aggregate(batch: _Batch, *, honor_parent_interrupt: bool = True
     results: list = []
     if len(batch.children) == 1:
         results.append(batch.run_child(*batch.children[0]))
+        # A one-child unit has no join to wait on, but everything after the child returns — host-owned finalize,
+        # transcript, manifest, then the durable write — is still owner-lifetime: record before any of it (#116000).
+        _record_finished_child(batch, results[-1], honor_parent_interrupt)
     else:
         _run_children_parallel(batch, results, honor_parent_interrupt=honor_parent_interrupt)