"""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.<op>(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 ``<label: ...exc>`` 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: <error: ", _prompt) + _diag_section("tool_schema: <error: ", _tools)

def _diag_threads(worker_thread: Optional[threading.Thread]) -> 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 ["  <worker frame not available>"])
    else:
        lines.append("  <no worker thread handle>" if worker_thread is None else "  <worker thread already exited>")

    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("<all-thread dump failed: ", _all_threads)

def _dump_subagent_timeout_diagnostic(
    *, child: Any, task_index: int, timeout_seconds: float, duration_seconds: float,
    worker_thread: Optional[threading.Thread], goal: str,
) -> 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-<sid>-<ts>.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}: <unreadable>"

        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(
            "<get_activity_summary failed: ", lambda: [f"  {k}: {v!r}" for k, v in child.get_activity_summary().items()],
        )
        lines += ["", *_diag_threads(worker_thread), "", "## Notes",
            "  This file is written ONLY when a subagent times out with 0 API calls.",
            "  0-API-call timeouts mean the child never reached its first LLM request.",
            "  Common causes: oversized prompt rejected by provider, transport hang,",
            "  credential resolution stuck. See issue #14726 for context.",
        ]
        dump_path.write_text("\n".join(lines), encoding="utf-8")
        return str(dump_path)
    except Exception as exc:
        logger.warning("Subagent timeout diagnostic dump failed: %s", exc)
        return None

# ── Per-run helpers ──────────────────────────────────────────────────────────

# Granularity for re-checking a child's progress while a configured ``child_timeout_seconds`` budget runs.
# Five seconds is finer than the 30s heartbeat and cheap (one activity-summary read per slice).
_LIVENESS_POLL_SECONDS = 5.0
# Fraction of the inactivity budget after which the child is warned once (via its steer channel) that the window is
# closing, so a child that is merely slow can wrap up and return instead of losing its whole context (#116001).
_BUDGET_WARNING_FRACTION = 0.8

def _budget_warning_text(idle_seconds: float, child_timeout: float) -> 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 "<none>",
                )
                # 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})