diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 61d3e3899e..648fc0e260 100644 --- a/tests/tools/test_async_delegation.py +++ b/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) diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 9efa18af4f..5e7f75039c 100644 --- a/tools/async_delegation.py +++ b/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)) diff --git a/tools/delegate_tool_dispatch.py b/tools/delegate_tool_dispatch.py index 6d21dc566a..8fe142ecbc 100644 --- a/tools/delegate_tool_dispatch.py +++ b/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 (``-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//; 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"]