"""Running ONE pre-built child agent: heartbeat, registry entry, workspace seeding, timeout/failure handling, result-entry assembly and cleanup (``_ChildRun``).""" from __future__ import annotations import logging import contextvars import json import os import threading import time from concurrent.futures import TimeoutError as FuturesTimeoutError from typing import Any, Dict, List, Optional from agent.compression_marker import elide from agent.interrupt_compat import request_hard_interrupt from dataclasses import dataclass, field from tools import file_state from tools.delegate_tool_progress import _quiet, _safe_progress from tools.delegate_tool_registry import ( _capture_gateway_steer_authority, _close_subagent_steering, _register_subagent, _unregister_subagent, ) from tools.delegate_tool_results import ( _extract_output_tail, _looks_like_error_output, _stringify_tool_content, _summarize_tool_arguments, ) logger = logging.getLogger("tools.delegate_tool") # log-record parity with the origin module def _num(value: Any, default: int = 0) -> int: """int() for counters that may be mocks/None on test doubles.""" return int(value) if isinstance(value, (int, float)) else default def _str_or_none(value: Any) -> Optional[str]: return value if isinstance(value, str) else None def _fabricated_entry(idx: int, status: str, error: str, child: Any, duration: float = 0) -> Dict[str, Any]: """Result entry for a child that raised, never finished, or was abandoned.""" return { "task_index": idx, "status": status, "summary": None, "error": error, "api_calls": 0, "duration_seconds": duration, "_child_role": getattr(child, "_delegate_role", None), } def _append_missed_steer(entry: Dict[str, Any], late_steer: Optional[str]) -> None: """Record steer text that won the race with the child's failure/timeout.""" if late_steer: entry["missed_steer"] = late_steer entry["error"] += (" [steer did not land before the subagent stopped: " f"{late_steer}]") def _close_child(child: Any, log_message: str) -> None: """Best-effort ``child.close()`` (tool sandboxes, browser daemons, httpx clients).""" with _quiet(log_message, exc_info=True): close = getattr(child, "close", None) if callable(close): close() def _with_children_lock(parent_agent: Any, op: str, child: Any) -> None: """``parent_agent._active_children.(child)`` under the parent's lock when it has one.""" lock = getattr(parent_agent, "_active_children_lock", None) if lock: with lock: getattr(parent_agent._active_children, op)(child) else: getattr(parent_agent._active_children, op)(child) def _attach_child(parent_agent: Any, child: Any) -> None: """Register the child for parent interrupt propagation. ``interrupt()`` fans out to a SNAPSHOT of ``_active_children``; a child attached after the stop landed (a fan-out still building its siblings, a turn that has not reached its iteration check yet) would otherwise start with no signal and run to completion as an orphan. Mirror a pending stop here so the whole spawn tree dies with its parent.""" if hasattr(parent_agent, "_active_children"): _with_children_lock(parent_agent, "append", child) if getattr(parent_agent, "_interrupt_requested", False) is not True: return # Same soft/hard split as ``interrupt()``'s own fan-out: a hard stop cancels, a soft one redirects. message = getattr(parent_agent, "_interrupt_message", None) hard = getattr(parent_agent, "_hard_interrupt_requested", None) if hard is None or hard.is_set(): _signal_child_stop(child, message or "parent agent interrupted", tool_reason=getattr(parent_agent, "_tool_interrupt_reason", None) or "parent agent interrupted") else: with _quiet("Failed to propagate interrupt to late child: %s"): child.interrupt(message) def _detach_child(parent_agent: Any, child: Any) -> None: """Remove the child from parent interrupt propagation (no-op if absent).""" if not hasattr(parent_agent, "_active_children"): return try: _with_children_lock(parent_agent, "remove", child) except (ValueError, UnboundLocalError) as e: logger.debug("Could not remove child from active_children: %s", e) def _signal_child_stop(child: Any, *reason: str, tool_reason: str = "parent delegation ended") -> None: """Cooperative interrupt so the child's worker thread can exit cleanly. ``tool_reason`` is the fixed cause the child's tools see (a pending approval wait reports it instead of a user deny).""" with _quiet(None): if (child is not None and not request_hard_interrupt(child, *reason, tool_reason=tool_reason) and hasattr(child, "_interrupt_requested")): child._interrupt_requested = True # ── 0-API-call timeout diagnostic ──────────────────────────────────────────── def _format_thread_stack(frame: Any, indent: str) -> List[str]: import traceback as _traceback return [f"{indent}{sub}" for frame_line in _traceback.format_stack(frame) for sub in frame_line.rstrip().split("\n")] _DIAG_CHILD_ATTRS = ( "model", "provider", "api_mode", "base_url", "max_iterations", "quiet_mode", "skip_memory", "skip_context_files", "platform", "_delegate_role", "_delegate_depth", ) def _diag_section(label: str, produce) -> List[str]: """Lines from ``produce()``, or one ```` line so a broken attribute never aborts the dump.""" try: return list(produce()) except Exception as exc: return [f" {label}{exc}>"] def _diag_sizes(child: Any) -> List[str]: def _prompt(): sys_prompt = getattr(child, "ephemeral_system_prompt", None) or getattr(child, "system_prompt", None) or "" is_str = isinstance(sys_prompt, str) return [ f" system_prompt_bytes: {len(sys_prompt.encode('utf-8')) if is_str else 'n/a'}", f" system_prompt_chars: {len(sys_prompt) if is_str else 'n/a'}", ] def _tools(): tools_schema = getattr(child, "tools", None) if tools_schema is None: return [] return [ f" tool_schema_count: {len(tools_schema)}", f" tool_schema_bytes: {len(json.dumps(tools_schema, default=str).encode('utf-8'))}", ] return ["## Prompt / schema sizes"] + _diag_section("system_prompt: List[str]: """Worker stack plus all other live threads (bounded to 40): the worker is often parked on a helper thread, so a pre-HTTP wedge is indistinguishable from a slow provider without the full picture.""" import sys as _sys lines = ["## Worker thread stack at timeout"] frames = _sys._current_frames() if worker_thread is not None and worker_thread.is_alive(): worker_frame = frames.get(worker_thread.ident) lines.extend(_format_thread_stack(worker_frame, " ") if worker_frame is not None else [" "]) else: lines.append(" " if worker_thread is None else " ") def _all_threads(): frames = _sys._current_frames() by_ident = {th.ident: th for th in threading.enumerate() if th.ident} worker_ident = worker_thread.ident if worker_thread else None out: List[str] = [] for dumped, (ident, frame) in enumerate(f for f in frames.items() if f[0] != worker_ident): # worker dumped above if dumped >= 40: out.append(f" <{len(frames) - dumped - 1} more threads omitted>") break th = by_ident.get(ident) out.append(f" --- {th.name if th else f'ident={ident}'}{' daemon' if (th and th.daemon) else ''} ---") out.extend(_format_thread_stack(frame, " ")) return out return lines + ["", "## All thread stacks at timeout"] + _diag_section(" Optional[str]: """Structured diagnostic for a subagent that timed out before any API call (otherwise "timed out with no response", 0 API calls, nothing to inspect): ``~/.hermes/logs/subagent-timeout--.log`` with the child's config, prompt/schema sizes, activity snapshot and worker stack. Path, or None on failure.""" try: from hermes_constants import get_hermes_home import datetime as _dt logs_dir = get_hermes_home() / "logs" try: logs_dir.mkdir(parents=True, exist_ok=True) except Exception: return None subagent_id = getattr(child, "_subagent_id", None) or f"idx{task_index}" dump_path = logs_dir / f"subagent-timeout-{subagent_id}-{_dt.datetime.now().strftime('%Y%m%d_%H%M%S')}.log" _goal_preview = elide((goal or "").strip(), 1000) def _attr_line(attr): try: return f" {attr}: {getattr(child, attr, None)!r}" except Exception: return f" {attr}: " tool_names = getattr(child, "valid_tool_names", None) lines: List[str] = [ "# Subagent timeout diagnostic — issue #14726", f"# Generated: {_dt.datetime.now().isoformat()}", "", "## Timeout", f" task_index: {task_index}", f" subagent_id: {subagent_id}", f" configured_timeout: {timeout_seconds}s", f" actual_duration: {duration_seconds:.2f}s", "", "## Goal", _goal_preview or "(empty)", "", "## Child config", *map(_attr_line, _DIAG_CHILD_ATTRS), "", "## Toolsets", f" enabled_toolsets: {getattr(child, 'enabled_toolsets', None)!r}", ] if tool_names: lines.append(f" loaded tool count: {len(tool_names)}") with _quiet(None): lines.append(f" loaded tools: {sorted(tool_names)}") lines += ["", *_diag_sizes(child), "", "## Activity summary"] lines += _diag_section( " str: return ( f"[delegation budget warning] No progress signal for {idle_seconds:.0f}s of your {child_timeout:.0f}s " "inactivity window. Finish the current step and return your summary now — the work is discarded if the " "window elapses." ) def _warn_child_budget(child: Any, idle_seconds: float, child_timeout: float) -> None: """Queue the one-line warning through the child's steer path (delivered at its next iteration boundary).""" steer = getattr(child, "steer", None) if not callable(steer): return try: steer(_budget_warning_text(idle_seconds, child_timeout)) except Exception as exc: logger.debug("budget warning steer failed: %s", exc) def _child_activity_fingerprint(child: Any) -> tuple: """``(completed calls, current tool, activity clock)`` — the progress signals the heartbeat's stale verdict already watches. A child waiting on an in-flight LLM completion is ALIVE: its activity clock ticks while the provider works (``direct_api_call``'s 15s heartbeat) and the call itself is bounded by the per-call stale watchdog, so at the delegation layer it must not read as stuck (#116001).""" try: summary = child.get_activity_summary() or {} except Exception: return (None, None, None) return (summary.get("api_call_count"), summary.get("current_tool"), summary.get("last_activity_ts")) def _child_last_event_age(child: Any) -> Optional[float]: """Seconds since the child's activity clock last ticked, or None when the child exposes no clock. Reported on a timeout so an operator can tell a slow-but-live provider from a runaway without transcript forensics.""" try: ts = (child.get_activity_summary() or {}).get("last_activity_ts") return round(max(0.0, time.time() - float(ts)), 2) if ts is not None else None except Exception: return None class _Heartbeat: """One child's parent-activity heartbeat via the shared periodic scheduler (``agent.periodic_scheduler``) — not one daemon thread per child. NOT started at construction: the caller calls ``start()`` inside its ``try`` so a failed schedule (OS thread exhaustion on first use) leaves ``handle`` None and ``stop()`` is a no-op.""" def __init__(self, child: Any, parent_agent: Any, task_index: int): self.child, self.parent_agent, self.task_index = child, parent_agent, task_index # Stale detection: a cycle counts as stale when (tool, iteration, # activity_ts) all froze; thresholds differ idle vs in-tool. self.last_seen = {"iter": 0, "tool": None, "ts": None, "stale": 0} self.handle = None # Set on the stale verdict; ``await_child`` waits on it (its worker's done-callback # sets it too) so a wedged child ends the wait instead of only ending the heartbeat. self.settled = threading.Event() # Threshold (seconds of frozen activity) the stale verdict fired at; None until it does. # ``await_child`` reads it to name the real cause when a configured cap was still pending. self.stale_threshold_seconds: Optional[float] = None def start(self) -> None: from agent.periodic_scheduler import schedule from tools.delegate_tool import _HEARTBEAT_INTERVAL self.handle = schedule(self.tick, _HEARTBEAT_INTERVAL) def stop(self) -> None: """wait=5 mirrors the old thread join: an in-flight tick finishes.""" if self.handle is not None: self.handle.cancel(wait=5) def tick(self): """Returning False stops the periodic callback.""" from tools.delegate_tool import _HEARTBEAT_INTERVAL, _HEARTBEAT_STALE_CYCLES_IDLE, _HEARTBEAT_STALE_CYCLES_IN_TOOL child, parent_agent, task_index, last_seen = self.child, self.parent_agent, self.task_index, self.last_seen touch = getattr(parent_agent, "_touch_activity", None) if parent_agent is not None else None if not touch: return None desc = f"delegate_task: subagent {task_index} working" try: child_summary = child.get_activity_summary() child_tool = child_summary.get("current_tool") child_iter = child_summary.get("api_call_count", 0) child_max = child_summary.get("max_iterations", 0) child_activity_ts = child_summary.get("last_activity_ts") # A slow model wait refreshes last_activity_ts (direct_api_call # heartbeat), so it never looks stale at the idle threshold. activity_advanced = child_activity_ts is not None and ( last_seen["ts"] is None or child_activity_ts > last_seen["ts"] ) if child_iter > last_seen["iter"] or child_tool != last_seen["tool"] or activity_advanced: last_seen.update(iter=child_iter, tool=child_tool, stale=0) if child_activity_ts is not None: last_seen["ts"] = child_activity_ts else: last_seen["stale"] += 1 stale_cycles = _HEARTBEAT_STALE_CYCLES_IN_TOOL if child_tool else _HEARTBEAT_STALE_CYCLES_IDLE if last_seen["stale"] >= stale_cycles: logger.warning( "Subagent %d appears stale (no progress for %d heartbeat cycles, tool=%s) — abandoning its wait", task_index, last_seen["stale"], child_tool or "", ) # A finite/-Q turn has no gateway watchdog behind this; the wait itself must end (#109749). self.stale_threshold_seconds = stale_cycles * _HEARTBEAT_INTERVAL self.settled.set() return False if child_tool: desc = f"delegate_task: subagent running {child_tool} (iteration {child_iter}/{child_max})" elif child_summary.get("last_activity_desc", ""): desc = f"delegate_task: subagent {child_summary.get('last_activity_desc', '')} (iteration {child_iter}/{child_max})" except Exception: pass with _quiet(None): touch(desc) return None def _start_heartbeat(child: Any, parent_agent: Any, task_index: int) -> _Heartbeat: """Build (not start) one child's heartbeat; see ``_Heartbeat``.""" return _Heartbeat(child, parent_agent, task_index) def _register_child( child: Any, parent_agent: Any, goal: str, *, owner_session_id: Optional[str], owner_transport: Any, owner_session_record: Any, ) -> Optional[str]: """Register the live child in the module registry; return its subagent_id. Test doubles without a stable string ``_subagent_id`` are not registered (None) and the caller skips every registry interaction for them.""" _subagent_id = getattr(child, "_subagent_id", None) if not isinstance(_subagent_id, str) or not _subagent_id: return None if owner_session_id is None: with _quiet(None): from gateway.session_context import get_session_env owner_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or None if owner_session_id and (owner_transport is None or owner_session_record is None): owner_transport, owner_session_record = _capture_gateway_steer_authority(owner_session_id) _raw_depth = getattr(child, "_delegate_depth", 1) _register_subagent({ "subagent_id": _subagent_id, "parent_id": _str_or_none(getattr(child, "_parent_subagent_id", None)), "depth": max(0, _raw_depth - 1) if isinstance(_raw_depth, int) else 0, "goal": goal, "delegation_id": _str_or_none(getattr(child, "_delegation_id", None)), "model": _str_or_none(getattr(child, "model", None)), "started_at": time.time(), "status": "running", "tool_count": 0, "agent": child, # Owning conversation's durable session id (same lineage completion delivery routes by), sourced from the # child's stamp so it survives a parent_agent rebuild between dispatch and run; used for list/steer/stop # ownership when the weakref chain breaks. "owner_agent_session_id": ( str(getattr(child, "_parent_session_id", "") or "") or str(getattr(parent_agent, "session_id", "") or "") or None ), # Immutable live gateway/TUI session that commissioned this child. # Empty outside those hosts; RPC authority fails closed. "owner_session_id": owner_session_id, "owner_transport": owner_transport, "owner_session_record": owner_session_record, }) return _subagent_id def _create_isolated_worktree(parent_agent: Any, parent_task_id: Any, subagent_id: Optional[str]): """Opt-in worktree isolation: own git worktree off the parent's HEAD (the child's terminal starts there). Git-only, local-backend-only; failures degrade silently to the shared workspace. Returns the worktree info or None.""" from tools.delegate_tool import _get_worktree_isolation, _resolve_workspace_hint if not _get_worktree_isolation(): return None with _quiet("worktree isolation setup failed: %s"): from tools import subagent_worktree if not subagent_worktree.local_backend_active(): logger.debug("worktree isolation skipped: non-local terminal backend") return None _parent_cwd = None with _quiet(None): from tools.terminal_tool import get_session_cwd as _gsc _parent_cwd = _gsc(parent_task_id) return subagent_worktree.create_subagent_worktree( _parent_cwd or _resolve_workspace_hint(parent_agent), subagent_id=subagent_id, ) return None def _defer_close_after_timeout(child: Any, child_future: Any) -> None: """Hand ``child.close()`` to a Future done-callback and drain its transports. The interrupt is cooperative: the worker still runs its finally path, so closing now could close SQLite under its final write — the done-callback is the first safe boundary. The abandoned worker is usually parked in an OpenSSL read; NEVER hard-close that transport from this thread (cross-thread FD release under a live SSL read corrupts native state) — shutdown() the pooled sockets so the read settles with EOF and the worker unwinds. One immediate sweep + one delayed re-sweep for a connection opened in between; a worker that still won't settle keeps its resources until process exit. """ child_future.add_done_callback(lambda _done: _close_child(child, "Failed to close timed-out child after worker exit")) # Bounded drain (#94248 native half): the deferred close above only fires once the abandoned worker # unwinds, but that worker is typically parked inside an in-flight OpenSSL read (Codex / httpx). Never # hard-close that transport from this thread — releasing FDs under a live SSL read is the #29507/#70773 # native-corruption family. Instead shutdown() the child's pooled sockets, which is FD-safe from any # thread and settles the blocked read with EOF/EPIPE so the worker can unwind and trigger the deferred # close. _drain = getattr(child, "_drain_transports_after_abandonment", None) if not callable(_drain): return def _drain_once(phase: str) -> None: with _quiet("Timed-out child transport drain (%s) failed", phase, exc_info=True): _drain(reason=f"delegate_timeout_{phase}") _drain_once("immediate") _resweep_timer = threading.Timer(5.0, lambda: None if child_future.done() else _drain_once("resweep")) _resweep_timer.daemon = True _resweep_timer.start() def _lease_child_credential(child: Any) -> tuple[Any, Optional[str]]: """Lease a credential from the child's pool (if any) and bind it; ``(pool, lease_id)``. The bound entry must serve the child's endpoint: on a mixed same-provider pool the least-leased pick may target another host, so it is released and an endpoint-matching entry is leased by id instead (#68237).""" child_pool = getattr(child, "_credential_pool", None) if child_pool is None: return None, None from agent.credential_pool import credential_pool_entry_serves_endpoint as _entry_serves_endpoint base_url = getattr(child, "base_url", None) leased_cred_id = child_pool.acquire_lease() if leased_cred_id is not None: with _quiet("Failed to bind child to leased credential: %s"): # Resolve the leased entry by id: the pool is shared with the parent/siblings, so current() is a # mutable cursor that may already point at someone else's pick. leased_entry = next((e for e in child_pool.entries() if e.id == leased_cred_id), None) if not _entry_serves_endpoint(leased_entry, base_url): child_pool.release_lease(leased_cred_id) leased_entry = next( (e for e in child_pool.entries() if e.last_status != "dead" and _entry_serves_endpoint(e, base_url)), None, ) leased_cred_id = child_pool.acquire_lease(leased_entry.id) if leased_entry is not None else None if leased_entry is not None and hasattr(child, "_swap_credential"): child._swap_credential(leased_entry) return child_pool, leased_cred_id def _merge_late_steer(result: Dict[str, Any], subagent_id: Optional[str], child: Any) -> None: """Linearization boundary for registry steering: from here the child cannot consume another steer. Closing under the registry lock either rejects a concurrent caller or drains every accepted exact text into the result before callbacks/result assembly run.""" late = _close_subagent_steering(subagent_id, child) if subagent_id else None if late: existing = result.get("pending_steer") result["pending_steer"] = f"{existing}\n{late}" if isinstance(existing, str) and existing else late @dataclass class _SchemaOutcome: schema: Optional[Dict[str, Any]] valid: Optional[bool] errors: List[str] retries: int def _validate_child_output_schema( child: Any, result: Dict[str, Any], task_index: int, child_task_id: str, relay_child_text: Any ) -> _SchemaOutcome: """Validate the final answer against the attached output_schema with ONE bounded retry. Schema-less children (no dict on ``child._delegate_output_schema``) take no branch here so their result entry stays byte-identical.""" _output_schema = getattr(child, "_delegate_output_schema", None) if not isinstance(_output_schema, dict): return _SchemaOutcome(_output_schema, None, [], 0) from tools.delegation_output_schema import build_retry_message, validate_output _first_text = result.get("final_response") or "" _schema_valid, _schema_errors = validate_output(_first_text, _output_schema) if _schema_valid or not _first_text.strip() or result.get("interrupted", False): return _SchemaOutcome(_output_schema, _schema_valid, _schema_errors, 0) # Exactly one retry turn, carrying the validation errors verbatim (no # schema re-paste — the child already holds the contract in its context). _retry_result = None try: # Same identity as the main child turn: this runs on the parent worker's thread, and an # unmarked turn is misread as the dispatcher-owned worker by every HERMES_KANBAN_* gate. from agent.delegation_context import delegated_child_context with delegated_child_context(str(getattr(child, "session_id", "") or "")): _retry_result = child.run_conversation( user_message=build_retry_message(_schema_errors), task_id=child_task_id, stream_callback=relay_child_text, ) except Exception as _retry_exc: logger.warning("Subagent %d schema-retry turn failed: %s", task_index, _retry_exc) if isinstance(_retry_result, dict): _retry_text = _retry_result.get("final_response") or "" if _retry_text.strip(): result["final_response"] = _retry_text try: result["api_calls"] = int(result.get("api_calls", 0) or 0) + int(_retry_result.get("api_calls", 0) or 0) except (TypeError, ValueError): pass _retry_messages = _retry_result.get("messages") if isinstance(_retry_messages, list) and isinstance(result.get("messages"), list): result["messages"] = result["messages"] + _retry_messages _schema_valid, _schema_errors = validate_output(_retry_text, _output_schema) return _SchemaOutcome(_output_schema, _schema_valid, _schema_errors, 1) def _build_tool_trace(messages: Any) -> list[Dict[str, Any]]: """Tool trace from the child's conversation messages, pairing parallel tool calls with their results by tool_call_id.""" tool_trace: list[Dict[str, Any]] = [] trace_by_id: Dict[str, Dict[str, Any]] = {} if not isinstance(messages, list): return tool_trace for msg in messages: if not isinstance(msg, dict): continue if msg.get("role") == "assistant": for tc in msg.get("tool_calls") or []: fn = tc.get("function", {}) arguments = fn.get("arguments", "") entry_t = { "tool": fn.get("name", "unknown"), "args_bytes": len(arguments), "input_summary": _summarize_tool_arguments(arguments), } tool_trace.append(entry_t) if tc.get("id"): trace_by_id[tc["id"]] = entry_t elif msg.get("role") == "tool": content = _stringify_tool_content(msg.get("content", "")) result_meta = {"result_bytes": len(content), "status": "error" if _looks_like_error_output(content) else "ok"} tc_id = msg.get("tool_call_id") target = trace_by_id.get(tc_id) if tc_id else None if target is not None: target.update(result_meta) elif tool_trace: tool_trace[-1].update(result_meta) # no tool_call_id: pair with the latest call return tool_trace def _build_result_entry( child: Any, result: Dict[str, Any], task_index: int, duration: float, schema: _SchemaOutcome, ) -> Dict[str, Any]: """Parent-visible result entry (status, exit_reason, tool trace, tokens, cost). ``status``/``exit_reason``/``truncated`` follow the ``_run_single_child`` contract; a structured failure always wins over the summary-presence heuristic (a fallback for legacy/mock results only).""" summary = result.get("final_response") or "" # "(empty)" is run_agent's give-up sentinel after repeated empty LLM # responses (usually a transport bug) — a failure, not a success. usable_summary = bool(summary) and summary.strip() != "(empty)" interrupt_note = "" if result.get("interrupted", False): status, exit_reason = "interrupted", "interrupted" # The loop's final_response is a placeholder here ("Operation interrupted…", also appended as the closing # assistant row); the completion must carry what the child actually had so far — its last real assistant # text — and keep the placeholder as the error. from agent.message_content import flatten_message_text placeholders = {"", summary.strip(), "Operation interrupted."} partial = next((t for m in reversed(result.get("messages") or []) if m.get("role") == "assistant" and (t := flatten_message_text(m.get("content")).strip()) not in placeholders), "") if partial: interrupt_note, summary = summary.strip(), partial elif result.get("failed") or result.get("error"): # The loop returns the error text as final_response, which would otherwise read as "completed". Never report a # provider rejection as "max_iterations" — that is only truthful for real budget exhaustion. status, exit_reason = "failed", "error" else: # exit_reason ("completed" vs "max_iterations") tells the parent HOW the task ended; completed=False with no # failure = budget exhaustion. A declared schema still violated after the bounded retry does NOT fail the # run: the child's raw final text is the deliverable (audits of up to 68 min were written off as "failed" # over a stray code fence or one missing field); ``schema_valid: false`` + ``schema_errors`` carry the # contract verdict, and the summary is prefixed with a notice so a status-only reader cannot mistake it # for validated output. exit_reason = "completed" if result.get("completed", False) else "max_iterations" status = "completed" if usable_summary else "failed" _cost = getattr(child, "session_estimated_cost_usd", 0.0) _cost_status = getattr(child, "session_cost_status", None) # Result entry contract: see the _run_single_child docstring. entry: Dict[str, Any] = { "task_index": task_index, "status": status, "summary": summary, "api_calls": result.get("api_calls", 0), "duration_seconds": duration, "model": _str_or_none(getattr(child, "model", None)), "exit_reason": exit_reason, # A budget-exhausted child still returns a summary (status stays # "completed"), so the parent needs this explicit flag. "truncated": exit_reason == "max_iterations", "tokens": { "input": _num(getattr(child, "session_prompt_tokens", 0)), "output": _num(getattr(child, "session_completion_tokens", 0)), }, "tool_trace": _build_tool_trace(result.get("messages") or []), # Captured before the finally block calls child.close() so the parent thread can fire subagent_stop with the # correct role; stripped before the dict is serialised back to the model (as is _child_cost_usd, folded into # the parent's session cost by the aggregator). "_child_role": getattr(child, "_delegate_role", None), "_child_cost_usd": float(_cost or 0.0) if isinstance(_cost, (int, float)) else 0.0, } # Model-visible per-delegation spend (unlike _child_cost_usd above). entry["cost_usd"] = round(entry["_child_cost_usd"], 6) entry["cost_status"] = _cost_status if isinstance(_cost_status, str) and _cost_status else "unknown" if status == "failed": entry["error"] = result.get("error", "Subagent did not produce a response.") # Classified reason from the child loop (e.g. "rate_limit", "billing") # lets the parent tell a quota wall from a task error without parsing prose. _failure_reason = result.get("failure_reason") if isinstance(_failure_reason, str) and _failure_reason: entry["failure_reason"] = _failure_reason elif interrupt_note: entry["error"] = interrupt_note # Schema-validation outcome — emitted ONLY when a schema was requested, so # legacy (schema-less) payloads keep their exact shape. if isinstance(schema.schema, dict): entry["schema_valid"] = bool(schema.valid) if schema.retries: entry["schema_retries"] = schema.retries if not schema.valid and schema.errors: entry["schema_errors"] = schema.errors if schema.valid is False and usable_summary: entry["schema_note"] = ( "Final answer does not satisfy the declared output_schema" + (" (after 1 retry)" if schema.retries else "") + "; `summary` is the child's raw, UNVALIDATED final text — extract what you need from it " "yourself (see schema_errors) rather than re-running the task." ) # A steer queued after the final assistant turn had no tool batch to land # in; name it so the parent sees it was MISSED rather than silently absorbed. _missed_steer = result.get("pending_steer") if isinstance(_missed_steer, str) and _missed_steer.strip(): entry["missed_steer"] = _missed_steer _miss_note = ("[steer did not land — the subagent finished before it could " f"be delivered: {_missed_steer}]") entry["summary"] = f"{summary}\n\n{_miss_note}" if summary else _miss_note return entry def _is_image_url(ref: str) -> bool: return ref.startswith(("http://", "https://", "data:image/")) def _build_child_goal_message(goal: str, images: List[str], child) -> Any: """The child's first user message when a task forwards ``images``. Routing reuses the inbound-image policy (``agent.image_routing``, honouring ``agent.image_input_mode``): a vision-capable child gets an OpenAI-style content list (text part + one ``image_url`` part per image; local files as data URLs behind the read guard, http(s)/data URLs verbatim); otherwise the goal gains ``[Image attached …]`` hint lines for ``vision_analyze``. Any failure degrades to the text-only goal so image plumbing never breaks a spawn — logged at warning since the caller asked for the images. """ try: # data: URLs ride as image parts only — their base64 never goes into the text hint or a text-mode goal. data_urls = [s for s in images if s.startswith("data:image/")] urls = [s for s in images if _is_image_url(s) and s not in data_urls] paths = [s for s in images if not _is_image_url(s)] from agent.image_routing import build_native_content_parts, decide_image_input_mode cfg = None with _quiet(None): from hermes_cli.config import load_config_readonly cfg = load_config_readonly() mode = decide_image_input_mode( str(getattr(child, "provider", "") or ""), str(getattr(child, "model", "") or ""), cfg, requested_provider=str(getattr(child, "requested_provider", "") or ""), ) if mode == "native": parts, skipped = build_native_content_parts(goal, paths, urls) if skipped: logger.warning("delegate_task: skipped %d unreadable image(s) for subagent: %s", len(skipped), ", ".join(skipped[:3])) if data_urls: parts = (parts or [{"type": "text", "text": goal}]) + [{"type": "image_url", "image_url": {"url": u}} for u in data_urls] return parts if any(p.get("type") == "image_url" for p in parts) else goal if data_urls: logger.warning("delegate_task: %d inline data-URL image(s) dropped for a non-vision subagent", len(data_urls)) hints: List[str] = [] for p in paths: if os.path.isfile(p): hints.append(f"[Image attached at: {p}]") else: logger.warning("delegate_task: image path not found, not forwarded: %s", p) hints.extend(f"[Image attached: {u}]" for u in urls) if not hints: return goal return goal + "\n\n" + "\n".join(hints) + "\nUse vision_analyze to inspect these images." except Exception: logger.warning("delegate_task: image forwarding failed; sending text-only goal", exc_info=True) return goal @dataclass class _ChildRun: """State of one child run, shared by every phase of ``_run_single_child``. ``worktree_info`` stays None until isolation engages (``attach_worktree`` is then a no-op on every early error path); ``seed_workspace`` sets the rest.""" child: Any parent_agent: Any task_index: int goal: str subagent_id: Optional[str] child_progress_cb: Any heartbeat: Any = None child_start: float = field(default_factory=time.monotonic) worktree_info: Optional[Dict[str, str]] = None child_task_id: str = "" parent_task_id: Optional[str] = None wall_start: float = 0.0 parent_reads_snapshot: list = field(default_factory=list) def elapsed(self) -> float: return round(time.monotonic() - self.child_start, 2) def relay_text(self, delta: str) -> None: """Stream callback forwarding the child's reply text up the progress relay so gateway watch windows mirror it live (subagent.text → message.delta). Inert under CLI/TUI: their progress handlers ignore non-tool events.""" if delta: _safe_progress(self.child_progress_cb, "subagent.text", preview=delta) def attach_worktree(self, entry_dict: Dict[str, Any]) -> Dict[str, Any]: """Inspect + prune the child worktree, reporting into the entry (no-op without isolation).""" info = self.worktree_info if info is None: return entry_dict from tools import subagent_worktree try: entry_dict["worktree"] = subagent_worktree.finalize_subagent_worktree(info) except Exception as e: # State is unknown: emit the SAME flagged schema the parent expects, # via the shared factory so the two producers never drift. logger.warning("worktree finalize failed: %s", e) entry_dict["worktree"] = subagent_worktree.unproven_worktree_payload(info, f"finalize raised: {e}") return entry_dict def seed_workspace(self) -> None: """Seed cwd/container aliases and optional worktree isolation for the child; ``goal`` is extended with the worktree contract note when isolation engaged.""" import uuid as _uuid self.child_task_id = self.subagent_id or f"subagent-{self.task_index}-{_uuid.uuid4().hex[:8]}" self.parent_task_id = getattr(self.parent_agent, "_current_task_id", None) # Seed the child's cwd record from the parent's: same starting directory, # but the child's later `cd`s stay in its own record. Per-session container # isolation keys containers by task_id; the child must share the PARENT's. with _quiet("Child cwd seed failed: %s"): from tools.terminal_tool import get_session_cwd, record_session_cwd, register_container_alias record_session_cwd(self.child_task_id, get_session_cwd(self.parent_task_id)) register_container_alias(self.child_task_id, self.parent_task_id) self.worktree_info = _create_isolated_worktree(self.parent_agent, self.parent_task_id, self.subagent_id) if self.worktree_info is not None: with _quiet("worktree cwd seed failed: %s"): from tools.terminal_tool import record_session_cwd as _rsc _rsc(self.child_task_id, self.worktree_info["path"]) # The child's context is already built; carry the isolation contract on # the goal message instead (same turn, no system-prompt mutation). from tools.subagent_worktree import build_worktree_context_note self.goal = self.goal + build_worktree_context_note(self.worktree_info) self.wall_start = time.time() self.parent_reads_snapshot = list(file_state.known_reads(self.parent_task_id)) if self.parent_task_id else [] def finish_failed( self, entry: Dict[str, Any], late_steer: Optional[str], *, preview: str, summary: str = "", status: Optional[str] = None, ) -> Dict[str, Any]: """Shared tail of every failure path: emit ``subagent.complete`` (``status`` defaults to the entry's), note the steer text that won the race with the failure, report the worktree.""" _safe_progress( self.child_progress_cb, "subagent.complete", preview=preview, status=status or entry["status"], duration_seconds=entry["duration_seconds"], summary=summary, ) _append_missed_steer(entry, late_steer) return self.attach_worktree(entry) def close_steering(self) -> Optional[str]: """Close steer acceptance (see ``_merge_late_steer``); returns late steer text, if any.""" return _close_subagent_steering(self.subagent_id, self.child) if self.subagent_id else None def wait_liveness_aware(self, settled: threading.Event, child_future: Any, child_timeout: Optional[float]) -> None: """Wait for the worker or the stale verdict, where ``child_timeout`` bounds INACTIVITY, not total runtime. A configured cap used to be a dispatch-to-death stopwatch, so it only ever killed children the runtime had already judged healthy: all 75 reported deaths happened while waiting on an in-flight LLM completion, none mid-tool (#116001). A provider serving multi-minute completions is progress — the child's activity clock ticks during the wait and the request is bounded by the per-call stale watchdog — so the budget restarts on every sign of progress, exactly the signals the heartbeat's stale verdict reads. Genuinely frozen children keep two authorities: this budget (when configured) and the heartbeat's stale threshold. """ if child_timeout is None: settled.wait() return deadline = time.monotonic() + child_timeout warn_at = deadline - child_timeout * (1.0 - _BUDGET_WARNING_FRACTION) warned = False fingerprint = _child_activity_fingerprint(self.child) while True: now = time.monotonic() remaining = deadline - now if remaining <= 0: return # no progress for the whole budget wait = min(_LIVENESS_POLL_SECONDS, remaining) if not warned: wait = min(wait, max(warn_at - now, 0.0)) settled.wait(timeout=wait) if settled.is_set(): return # the worker finished, or the heartbeat declared the child stale current = _child_activity_fingerprint(self.child) if current != fingerprint: fingerprint = current deadline = time.monotonic() + child_timeout warn_at = deadline - child_timeout * (1.0 - _BUDGET_WARNING_FRACTION) warned = False # a fresh window gets its own warning elif not warned and time.monotonic() >= warn_at: warned = True _warn_child_budget(self.child, child_timeout - (deadline - time.monotonic()), child_timeout) def await_child(self) -> tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]], bool]: """Run the child's conversation on a daemon worker: ``(result, None, False)`` or ``(None, error_entry, close_deferred)`` on timeout/exception. Hard timeout is off by default (``result(timeout=None)``; stuck children are the heartbeat's job). Daemon worker: an abandoned timed-out child on a non-daemon thread would block interpreter exit at atexit join. The worker installs a non-interactive approval callback (deny/approve per delegation.subagent_auto_approve) so dangerous-command prompts never fall back to ``input()`` and deadlock the parent TUI. On failure: steer acceptance closes BEFORE the stop signal (a concurrent steer is drained into the entry or rejected, never lost); a 0-API-call timeout gets a diagnostic dump; a worker that still owns the child gets ``child.close()`` via a Future done-callback (``close_deferred=True``) — closing here would race its still-unwinding finally path. """ from tools.delegate_tool import (_get_child_timeout, _get_subagent_approval_callback, _set_subagent_approval_cb) from tools.daemon_pool import DaemonThreadPoolExecutor child, task_index = self.child, self.task_index child_timeout = _get_child_timeout() executor = DaemonThreadPoolExecutor( max_workers=1, initializer=_set_subagent_approval_cb, initargs=(_get_subagent_approval_callback(),), ) # Worker thread handle so the timeout diagnostic can dump its stack. worker_thread_holder: Dict[str, Optional[threading.Thread]] = {"t": None} # Resolved after seed_workspace so a multimodal goal's text part carries the worktree note too. _images = list(getattr(child, "_delegate_images", None) or []) user_message: Any = _build_child_goal_message(self.goal, _images, child) if _images else self.goal def _run_with_thread_capture(): worker_thread_holder["t"] = threading.current_thread() from agent.delegation_context import delegated_child_context with delegated_child_context(str(getattr(child, "session_id", "") or "")): return child.run_conversation( user_message=user_message, task_id=self.child_task_id, stream_callback=self.relay_text, ) future = executor.submit(contextvars.copy_context().run, _run_with_thread_capture) # One wait covers both ways out: the worker finishing, or the heartbeat's stale verdict. # Without the second, a worker wedged after its final answer holds a finite (-Q / Bot Chat # one-shot) turn — and its session lease — forever, since that runtime has no gateway # inactivity watchdog (#109749). The configured cap bounds INACTIVITY (see wait_liveness_aware). settled = self.heartbeat.settled if self.heartbeat is not None else threading.Event() future.add_done_callback(lambda _f: settled.set()) # Set when the stale verdict — not the configured cap — ended the wait; the entry must name that cause. stale_after: Optional[float] = None try: self.wait_liveness_aware(settled, future, child_timeout) if not future.done(): stale_after = getattr(self.heartbeat, "stale_threshold_seconds", None) raise FuturesTimeoutError() return future.result(), None, False except Exception as wait_exc: exc: BaseException = wait_exc # ``as`` targets are unbound after the except block finally: # Shut down without waiting — a child stuck on blocking I/O would hang wait=True forever. executor.shutdown(wait=False) _late_pending_steer = self.close_steering() _signal_child_stop(child) is_timeout = isinstance(exc, (FuturesTimeoutError, TimeoutError)) # What actually ended the wait: the stale threshold pre-empts a longer configured cap. timeout_cause = stale_after if stale_after is not None else child_timeout duration = self.elapsed() logger.warning("Subagent %d %s after %.1fs", task_index, "timed out" if is_timeout else f"raised {type(exc).__name__}", duration) child_api_calls = 0 with _quiet(None): child_api_calls = int(child.get_activity_summary().get("api_call_count", 0) or 0) # A timeout BEFORE any API call is a black box without a diagnostic dump. before_first_call = is_timeout and child_api_calls == 0 diagnostic_path: Optional[str] = None if before_first_call: diagnostic_path = _dump_subagent_timeout_diagnostic( child=child, task_index=task_index, # A stale verdict or a configured cap; ``or 0.0`` guards the type checker. timeout_seconds=float(timeout_cause or 0.0), duration_seconds=float(duration), worker_thread=worker_thread_holder.get("t"), goal=self.goal, ) if diagnostic_path: logger.warning("Subagent %d 0-API-call timeout — diagnostic written to %s", task_index, diagnostic_path) if not is_timeout: _err = str(exc) elif stale_after is not None: _err = ( f"Subagent stopped making progress after {child_api_calls} API call(s) — no activity for " f"{stale_after:g}s (heartbeat stale threshold); the pending worker was abandoned." ) elif before_first_call: _err = ( f"Subagent timed out after {child_timeout}s without making any API call — the child never reached its " f"first LLM request (prompt construction, credential resolution, or transport may be stuck)." ) else: _err = ( f"Subagent timed out after {child_timeout}s with {child_api_calls} API call(s) completed — no " f"progress in that window (no completed call, tool change, or activity-clock tick): a stalled " f"provider request or an unresponsive network request." ) if diagnostic_path: _err += f" Diagnostic: {diagnostic_path}" status = "timeout" if is_timeout else "error" _error_entry = { "task_index": task_index, "status": status, "summary": None, "error": _err, "exit_reason": status, "api_calls": child_api_calls, "duration_seconds": duration, "timeout_seconds": timeout_cause if is_timeout else None, "timed_out_after_seconds": duration if is_timeout else None, "timeout_phase": "before_first_llm_call" if before_first_call else "after_llm_calls" if is_timeout else None, # How long the child had been silent when the wait ended: distinguishes a slow provider from a runaway # without transcript forensics (#116001). "last_event_age": _child_last_event_age(child) if is_timeout else None, "_child_role": getattr(child, "_delegate_role", None), "diagnostic_path": diagnostic_path, } self.finish_failed(_error_entry, _late_pending_steer, preview=f"Timed out after {duration}s" if is_timeout else str(exc)) close_deferred = is_timeout and not future.done() if close_deferred: _defer_close_after_timeout(child, future) return None, _error_entry, close_deferred def append_sibling_write_reminder(self, entry: Dict[str, Any]) -> None: """Warn the parent when this child wrote files the parent had already read. Checks writes by ANY non-parent task_id (not just this child's) so nested orchestrator→worker chains are covered too.""" if not (self.parent_task_id and self.parent_reads_snapshot): return with _quiet("file_state sibling-write check failed", exc_info=True): sibling_writes = file_state.writes_since(self.parent_task_id, self.wall_start, self.parent_reads_snapshot) mod_paths = sorted({p for paths in sibling_writes.values() for p in paths}) if sibling_writes else [] if not mod_paths: return reminder = ( "\n\n[NOTE: subagent modified files the parent previously read — re-read before editing: " + ", ".join(mod_paths[:8]) + (f" (+{len(mod_paths) - 8} more)" if len(mod_paths) > 8 else "") + "]" ) if entry.get("summary"): entry["summary"] = entry["summary"] + reminder else: entry["stale_paths"] = mod_paths def account_background_processes(self, entry: Dict[str, Any]) -> None: """Name the child's background processes on the result BEFORE ``cleanup`` kills them: handed-off ones now belong to the parent (their completion lands in the parent's chat); anything else still running is about to be terminated, and the parent must hear that from the runtime rather than trust a child's "watcher running".""" handed = list(getattr(self.child, "_handed_off_processes", None) or []) if handed: entry["handed_off_processes"] = handed with _quiet(None): from tools.process_registry import process_registry, _output_tail leftover = process_registry.running_owned_by(self.child_task_id) if leftover: entry["orphaned_processes"] = [ {"session_id": s.id, "command": s.command[:200], "runtime_seconds": round(time.time() - s.started_at)} for s in leftover] unread = process_registry.unread_completions_owned_by(self.child_task_id) if unread: entry["unread_completions"] = [ {"session_id": s.id, "command": s.command[:200], "exit_code": s.exit_code, "output_tail": _output_tail(s, 600)} for s in unread] def emit_complete(self, result: Dict[str, Any], entry: Dict[str, Any], duration: float) -> None: """Fire ``subagent.complete`` with the per-branch observability payload (tokens, cost, files touched, tool-output tail); every field is optional and degrades gracefully on the client.""" if not self.child_progress_cb: return child = self.child summary = entry["summary"] _files_read: list = [] with _quiet(None): _files_read = list(file_state.known_reads(self.child_task_id))[:40] _files_written_map: dict = {} with _quiet(None): _files_written_map = file_state.writes_since("", self.wall_start, []) # all writes since wall_start complete_kwargs: Dict[str, Any] = { "preview": summary[:160] if summary else entry.get("error", ""), "status": entry["status"], "duration_seconds": duration, "summary": summary[:500] if summary else entry.get("error", ""), "input_tokens": _num(getattr(child, "session_prompt_tokens", 0)), "output_tokens": _num(getattr(child, "session_completion_tokens", 0)), "reasoning_tokens": _num(getattr(child, "session_reasoning_tokens", 0)), "api_calls": _num(entry["api_calls"]), "files_read": _files_read, "files_written": sorted({p for tid, paths in _files_written_map.items() if tid == self.child_task_id for p in paths})[:40], "output_tail": _extract_output_tail(result, max_entries=8, max_chars=600), } if entry.get("failure_reason"): # Classified verdict rides the event so every surface glosses the failure the same way. complete_kwargs["failure_reason"] = entry["failure_reason"] _cost_usd = getattr(child, "session_estimated_cost_usd", None) if _cost_usd is not None: with _quiet(None): complete_kwargs["cost_usd"] = float(_cost_usd) _safe_progress(self.child_progress_cb, "subagent.complete", **complete_kwargs) def cleanup(self, *, heartbeat: _Heartbeat, child_pool: Any, leased_cred_id: Any, close_deferred: bool) -> None: """Finally-path teardown (idempotent, never raises). Order matters: stop heartbeat → drop registry entry → release credential lease → restore the parent's process-global tool names → detach from the parent's interrupt list → close the child (unless a timed-out worker still owns it) → pop the child's Relay scope if no turn is active.""" child = self.child heartbeat.stop() # Safe even if the child was never registered (ID missing on test doubles). if self.subagent_id: _unregister_subagent(self.subagent_id, agent=child) if child_pool is not None and leased_cred_id is not None: with _quiet("Failed to release credential lease: %s"): child_pool.release_lease(leased_cred_id) # Restore the parent's tool names so the process-global is correct for # any subsequent execute_code calls or other consumers. import model_tools saved_tool_names = getattr(child, "_delegate_saved_tool_names", None) if isinstance(saved_tool_names, list): model_tools._last_resolved_tool_names = list(saved_tool_names) _detach_child(self.parent_agent, child) # Close tool resources (terminal sandboxes, browser daemons, background # processes, httpx clients) so subagent subprocesses don't outlive the delegation. if not close_deferred: _close_child(child, "Failed to close child agent after delegation") # The child's execute_code kernels live exactly as long as the child (pinned against the LRU # cap while it runs); dispose them here so they never squat the cap after the child is gone. with _quiet("Failed to dispose child execute_code kernels: %s"): from tools.code_kernel import shutdown_kernels_for_delegated_child shutdown_kernels_for_delegated_child(str(getattr(child, "session_id", "") or "")) # The AIAgent turn boundary normally closes the child scope itself. This fallback covers failures before that # boundary starts, but must not pop a scope while a timed-out child worker is still unwinding. with _quiet("Failed to close child Relay session after delegation"): from agent import relay_runtime runtime = relay_runtime.get_runtime(create=False) child_session_id = str(getattr(child, "session_id", "") or "") child_turn_is_active = relay_runtime.SESSION_COORDINATOR.has_active_turn( profile_key=relay_runtime.current_profile_key(), session_id=child_session_id, ) if runtime is not None and child_session_id and not child_turn_is_active: runtime.unregister_subagent({"child_session_id": child_session_id})