fix subagent lifecycle ownership invariants

agent/subagent_lifecycle.py

@@ -7,14 +7,17 @@ sessions; plugins must obtain it from ``PluginContext.subagent_lifecycle``.
 
 from __future__ import annotations
 
+import contextvars
 import dataclasses
 import enum
 import hashlib
 import hmac
 import json
+import math
 import secrets
 import threading
 import time
+from contextlib import contextmanager
 from concurrent.futures import Future, ThreadPoolExecutor, TimeoutError
 from typing import Any, Callable, Mapping, Optional
 
@@ -155,6 +158,24 @@ class _Registry:
 _REGISTRY = _Registry()
 _EXECUTOR = ThreadPoolExecutor(max_workers=8, thread_name_prefix="hermes-lifecycle")
 _SECRET = secrets.token_bytes(32)
+_ACTIVE_PARENT_AGENT: contextvars.ContextVar[Any] = contextvars.ContextVar(
+    "hermes_subagent_lifecycle_parent", default=None
+)
+
+
+@contextmanager
+def bind_subagent_parent(parent_agent: Any):
+    """Bind the host-owned parent for the current agent turn."""
+    token = _ACTIVE_PARENT_AGENT.set(parent_agent)
+    try:
+        yield
+    finally:
+        _ACTIVE_PARENT_AGENT.reset(token)
+
+
+def get_active_subagent_parent() -> Any:
+    """Return the parent bound to this execution context, if any."""
+    return _ACTIVE_PARENT_AGENT.get()
 
 
 class SubagentLifecycleService:
@@ -190,9 +211,12 @@ class SubagentLifecycleService:
 
         # Delegate construction remains internal so plugin code never imports
         # private delegation helpers or manipulates the active-child registry.
-        from tools.delegate_tool import _build_child_agent, DEFAULT_MAX_ITERATIONS
+        from tools.delegate_tool import (
+            _build_child_preserving_parent_tools,
+            DEFAULT_MAX_ITERATIONS,
+        )
 
-        child = _build_child_agent(
+        child = _build_child_preserving_parent_tools(
             task_index=0,
             goal=request.goal,
             context=request.context,
@@ -311,9 +335,31 @@ class SubagentLifecycleService:
     def _record(self, handle: SubagentHandle) -> Optional[_Record]:
         if (
             not isinstance(handle, SubagentHandle)
+            or type(handle.contract_version) is not int
             or handle.contract_version != PUBLIC_CONTRACT_VERSION
         ):
             return None
+        if (
+            not isinstance(handle.subagent_id, str)
+            or not handle.subagent_id
+            or (
+                handle.parent_session_id is not None
+                and not isinstance(handle.parent_session_id, str)
+            )
+            or (
+                handle.correlation_id is not None
+                and not isinstance(handle.correlation_id, str)
+            )
+            or isinstance(handle.created_at, bool)
+            or not isinstance(handle.created_at, (int, float))
+            or not math.isfinite(handle.created_at)
+            or (handle.provider is not None and not isinstance(handle.provider, str))
+            or (handle.model is not None and not isinstance(handle.model, str))
+            or not isinstance(handle.role, str)
+            or type(handle.depth) is not int
+            or not isinstance(handle.capability, str)
+        ):
+            return None
         if not hmac.compare_digest(
             handle.capability,
             self._capability(
@@ -349,13 +395,14 @@ class SubagentLifecycleService:
 
     def _run(self, record: _Record, goal: str, parent: Any) -> None:
         with _REGISTRY.lock:
-            record.state = SubagentState.RUNNING
+            if record.state is not SubagentState.CANCEL_REQUESTED:
+                record.state = SubagentState.RUNNING
             record.started_at = time.time()
             record.updated_at = record.started_at
         try:
-            from tools.delegate_tool import _run_single_child
+            from tools.delegate_tool import _run_child_lifecycle
 
-            raw = _run_single_child(0, goal, record.agent, parent)
+            raw = _run_child_lifecycle(0, goal, record.agent, parent)
             status = (
                 str(raw.get("status", "error")) if isinstance(raw, dict) else "error"
             )

hermes_cli/plugins.py

@@ -374,10 +374,12 @@ class PluginContext:
         snapshots; they never receive a live agent or a private registry.
         """
         if self._subagent_lifecycle is None:
-            from agent.subagent_lifecycle import SubagentLifecycleService
+            from agent.subagent_lifecycle import (
+                SubagentLifecycleService,
+                get_active_subagent_parent,
+            )
             self._subagent_lifecycle = SubagentLifecycleService(
-                lambda: getattr(self._manager._cli_ref, "agent", None)
-                if self._manager._cli_ref is not None else None
+                get_active_subagent_parent
             )
         return self._subagent_lifecycle
 

run_agent.py

@@ -6822,6 +6822,8 @@ class AIAgent:
             reset_conversation_context,
             set_conversation_context,
         )
+        from agent.subagent_lifecycle import bind_subagent_parent
+
         # Publish the conversation id for ambient Nous Portal tagging. Every
         # LLM call made inside this turn — main loop, compression, vision,
         # web_extract, session_search, MoA slots, background-review forks
@@ -6841,7 +6843,7 @@ class AIAgent:
         # replaces the value with the live runtime after fallback restoration.
         # Keep the scope local instead of storing ContextVar tokens on the agent,
         # which may be observed from another thread.
-        with scoped_runtime_main({}):
+        with bind_subagent_parent(self), scoped_runtime_main({}):
             try:
                 return run_conversation(
                     self,

tests/agent/test_subagent_lifecycle.py

@@ -2,6 +2,7 @@
 
 import time
 from types import SimpleNamespace
+from unittest.mock import Mock
 
 import pytest
 
@@ -10,6 +11,8 @@ from agent.subagent_lifecycle import (
     SubagentLifecycleError,
     SubagentLifecycleService,
     SubagentState,
+    bind_subagent_parent,
+    get_active_subagent_parent,
 )
 
 
@@ -68,7 +71,7 @@ def test_launch_wait_result_and_handle_round_trip(lifecycle):
 
 
 def test_duplicate_correlation_and_permission_validation(lifecycle):
-    lifecycle.launch(SubagentLaunchRequest(goal="x", correlation_id="same"))
+    handle = lifecycle.launch(SubagentLaunchRequest(goal="x", correlation_id="same"))
     with pytest.raises(SubagentLifecycleError, match="Duplicate"):
         lifecycle.launch(SubagentLaunchRequest(goal="x", correlation_id="same"))
     with pytest.raises(SubagentLifecycleError, match="broaden"):
@@ -77,6 +80,7 @@ def test_duplicate_correlation_and_permission_validation(lifecycle):
         )
     with pytest.raises(SubagentLifecycleError, match="working_directory"):
         lifecycle.launch(SubagentLaunchRequest(goal="x", working_directory="C:/"))
+    lifecycle.wait(handle, timeout_seconds=1)
 
 
 def test_cancel_is_cooperative_and_forged_handle_is_unknown(lifecycle):
@@ -96,3 +100,153 @@ def test_simultaneous_launches_are_distinct_and_reconnect_is_in_process(lifecycl
     handles = [lifecycle.launch(SubagentLaunchRequest(goal="x")) for _ in range(10)]
     assert len({h.subagent_id for h in handles}) == 10
     assert lifecycle.reconnect(handles[0]).connected
+    for handle in handles:
+        lifecycle.wait(handle, timeout_seconds=1)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("capability", []),
+        ("contract_version", True),
+        ("subagent_id", None),
+        ("parent_session_id", []),
+        ("correlation_id", []),
+        ("created_at", "yesterday"),
+        ("provider", []),
+        ("model", []),
+        ("role", []),
+        ("depth", "one"),
+    ],
+)
+def test_malformed_deserialized_handle_is_unknown(lifecycle, field, value):
+    handle = lifecycle.launch(SubagentLaunchRequest(goal="x"))
+    malformed = handle.from_dict({**handle.to_dict(), field: value})
+
+    assert lifecycle.status(malformed).state is SubagentState.UNKNOWN
+    assert lifecycle.result(malformed).error_classification == "UNKNOWN_HANDLE"
+    lifecycle.wait(handle, timeout_seconds=1)
+
+
+def test_launch_preserves_parent_tool_resolution(monkeypatch):
+    import model_tools
+
+    parent = SimpleNamespace(session_id="parent-tools", enabled_toolsets=["file"])
+    model_tools._last_resolved_tool_names = ["parent_tool"]
+
+    def build(**_kwargs):
+        model_tools._last_resolved_tool_names = ["child_tool"]
+        return FakeChild("sa-tools")
+
+    monkeypatch.setattr("tools.delegate_tool._build_child_agent", build)
+    monkeypatch.setattr(
+        "tools.delegate_tool._run_single_child",
+        lambda *_args, **_kwargs: {
+            "status": "completed",
+            "summary": "done",
+            "api_calls": 0,
+            "duration_seconds": 0,
+        },
+    )
+
+    service = SubagentLifecycleService(lambda: parent)
+    handle = service.launch(SubagentLaunchRequest(goal="x"))
+
+    assert model_tools._last_resolved_tool_names == ["parent_tool"]
+    assert handle.subagent_id == "sa-tools"
+    service.wait(handle, timeout_seconds=1)
+
+
+def test_public_lifecycle_runs_host_aggregation(monkeypatch):
+    memory = Mock()
+    parent = SimpleNamespace(
+        session_id="parent-aggregate",
+        enabled_toolsets=["file"],
+        _memory_manager=memory,
+        _current_turn_id="turn-1",
+        session_estimated_cost_usd=1.0,
+        session_cost_source="none",
+        session_cost_status="unknown",
+    )
+    child = FakeChild("sa-aggregate")
+    child.session_id = "child-session"
+    hook = Mock()
+
+    monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **_kwargs: child)
+    monkeypatch.setattr(
+        "tools.delegate_tool._run_single_child",
+        lambda *_args, **_kwargs: {
+            "task_index": 0,
+            "status": "completed",
+            "summary": "aggregated",
+            "api_calls": 1,
+            "duration_seconds": 0.25,
+            "_child_role": "leaf",
+            "_child_cost_usd": 2.5,
+        },
+    )
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", hook)
+
+    service = SubagentLifecycleService(lambda: parent)
+    handle = service.launch(SubagentLaunchRequest(goal="aggregate me"))
+    assert service.wait(handle, timeout_seconds=1).state is SubagentState.SUCCEEDED
+
+    memory.on_delegation.assert_called_once_with(
+        task="aggregate me", result="aggregated", child_session_id="child-session"
+    )
+    hook.assert_called_once_with(
+        "subagent_stop",
+        parent_session_id="parent-aggregate",
+        parent_turn_id="turn-1",
+        child_session_id="child-session",
+        child_role="leaf",
+        child_summary="aggregated",
+        child_status="completed",
+        duration_ms=250,
+    )
+    assert parent.session_estimated_cost_usd == 3.5
+    assert parent.session_cost_source == "subagent"
+    assert parent.session_cost_status == "estimated"
+
+
+def test_plugin_context_uses_turn_scoped_parent(monkeypatch):
+    from hermes_cli.plugins import PluginContext, PluginManifest
+
+    parent = SimpleNamespace(session_id="gateway-parent", enabled_toolsets=["file"])
+    monkeypatch.setattr(
+        "tools.delegate_tool._build_child_agent", lambda **_kwargs: FakeChild("sa-gateway")
+    )
+    monkeypatch.setattr(
+        "tools.delegate_tool._run_single_child",
+        lambda *_args, **_kwargs: {
+            "status": "completed",
+            "summary": "done",
+            "api_calls": 0,
+            "duration_seconds": 0,
+        },
+    )
+    manager = SimpleNamespace(_cli_ref=None)
+    ctx = PluginContext(PluginManifest(name="test", source="test"), manager)
+
+    with bind_subagent_parent(parent):
+        handle = ctx.subagent_lifecycle.launch(SubagentLaunchRequest(goal="x"))
+        ctx.subagent_lifecycle.wait(handle, timeout_seconds=1)
+
+    assert handle.parent_session_id == "gateway-parent"
+
+
+def test_agent_turn_binds_and_clears_lifecycle_parent(monkeypatch):
+    from run_agent import AIAgent
+
+    agent = AIAgent.__new__(AIAgent)
+    observed = []
+
+    def run_conversation(parent, *_args, **_kwargs):
+        observed.append(get_active_subagent_parent())
+        return {"final_response": "ok"}
+
+    monkeypatch.setattr("agent.conversation_loop.run_conversation", run_conversation)
+
+    assert agent.run_conversation("hello") == {"final_response": "ok"}
+    assert observed == [agent]
+    assert get_active_subagent_parent() is None

tools/delegate_tool.py

@@ -2576,6 +2576,150 @@ def _run_single_child(
             logger.debug("Failed to close child agent after delegation")
 
 
+_PARENT_FINALIZATION_LOCK_GUARD = threading.Lock()
+_PARENT_FINALIZATION_FALLBACK_LOCK = threading.RLock()
+_CHILD_CONSTRUCTION_LOCK = threading.RLock()
+
+
+def _build_child_preserving_parent_tools(**kwargs):
+    """Build a child without leaking its resolved toolset into the parent."""
+    import model_tools
+
+    with _CHILD_CONSTRUCTION_LOCK:
+        parent_tool_names = list(model_tools._last_resolved_tool_names)
+        try:
+            child = _build_child_agent(**kwargs)
+        finally:
+            model_tools._last_resolved_tool_names = parent_tool_names
+    child._delegate_saved_tool_names = parent_tool_names
+    return child
+
+
+def _parent_finalization_lock(parent_agent) -> threading.RLock:
+    """Return the per-parent lock that serializes lifecycle side effects."""
+    if parent_agent is None:
+        return _PARENT_FINALIZATION_FALLBACK_LOCK
+    lock = getattr(parent_agent, "_subagent_finalization_lock", None)
+    if lock is not None:
+        return lock
+    with _PARENT_FINALIZATION_LOCK_GUARD:
+        lock = getattr(parent_agent, "_subagent_finalization_lock", None)
+        if lock is None:
+            lock = threading.RLock()
+            try:
+                setattr(parent_agent, "_subagent_finalization_lock", lock)
+            except Exception:
+                return _PARENT_FINALIZATION_FALLBACK_LOCK
+    return lock
+
+
+def _finalize_child_results(
+    results: List[Dict[str, Any]],
+    task_list: List[Dict[str, Any]],
+    children: List[tuple[int, Dict[str, Any], Any]],
+    parent_agent,
+) -> None:
+    """Apply host-owned summary, memory, hook, and cost contracts once."""
+    with _parent_finalization_lock(parent_agent):
+        _apply_summary_budget(results, parent_agent)
+        child_by_index = {index: child for index, _task, child in children}
+
+        if parent_agent and getattr(parent_agent, "_memory_manager", None):
+            for entry in results:
+                try:
+                    task_index = entry.get("task_index", -1)
+                    task_goal = (
+                        task_list[task_index]["goal"]
+                        if isinstance(task_index, int)
+                        and 0 <= task_index < len(task_list)
+                        else ""
+                    )
+                    child = child_by_index.get(task_index)
+                    parent_agent._memory_manager.on_delegation(
+                        task=task_goal,
+                        result=entry.get("summary", "") or "",
+                        child_session_id=getattr(child, "session_id", ""),
+                    )
+                except Exception:
+                    pass
+
+        parent_session_id = getattr(parent_agent, "session_id", None)
+        try:
+            from hermes_cli.plugins import invoke_hook as invoke_hook
+        except Exception:
+            invoke_hook = None
+
+        children_cost_total = 0.0
+        for entry in results:
+            child_role = entry.pop("_child_role", None)
+            child_cost = entry.pop("_child_cost_usd", 0.0)
+            try:
+                if child_cost:
+                    children_cost_total += float(child_cost)
+            except (TypeError, ValueError):
+                pass
+            if invoke_hook is None:
+                continue
+            try:
+                child_index = entry.get("task_index", -1)
+                child = child_by_index.get(child_index)
+                invoke_hook(
+                    "subagent_stop",
+                    parent_session_id=parent_session_id,
+                    parent_turn_id=getattr(parent_agent, "_current_turn_id", "") or "",
+                    child_session_id=getattr(child, "session_id", None),
+                    child_role=child_role,
+                    child_summary=entry.get("summary"),
+                    child_status=entry.get("status"),
+                    tool_call_history=_subagent_stop_tool_call_history(
+                        entry.get("tool_trace")
+                    ),
+                    duration_ms=int((entry.get("duration_seconds") or 0) * 1000),
+                )
+            except Exception:
+                logger.debug("subagent_stop hook invocation failed", exc_info=True)
+
+        if children_cost_total > 0.0:
+            try:
+                current = float(
+                    getattr(parent_agent, "session_estimated_cost_usd", 0.0) or 0.0
+                )
+                parent_agent.session_estimated_cost_usd = current + children_cost_total
+                if getattr(parent_agent, "session_cost_source", "none") in {
+                    None,
+                    "",
+                    "none",
+                }:
+                    parent_agent.session_cost_source = "subagent"
+                if getattr(parent_agent, "session_cost_status", "unknown") in {
+                    None,
+                    "",
+                    "unknown",
+                }:
+                    parent_agent.session_cost_status = "estimated"
+            except Exception:
+                logger.debug("Subagent cost rollup failed", exc_info=True)
+
+
+def _run_child_lifecycle(
+    task_index: int,
+    goal: str,
+    child=None,
+    parent_agent=None,
+) -> Dict[str, Any]:
+    """Run one child and apply the same host lifecycle used by delegate_task."""
+    result = _run_single_child(task_index, goal, child, parent_agent)
+    result.setdefault("task_index", task_index)
+    task = {"goal": goal}
+    _finalize_child_results(
+        [result],
+        [{"goal": ""} for _ in range(task_index)] + [task],
+        [(task_index, task, child)],
+        parent_agent,
+    )
+    return result
+
+
 def _recover_tasks_from_json_string(
     tasks: Any,
 ) -> tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
@@ -2746,13 +2890,6 @@ def delegate_task(
         task_list, context
     )
 
-    # Save parent tool names BEFORE any child construction mutates the global.
-    # _build_child_agent() calls AIAgent() which calls get_tool_definitions(),
-    # which overwrites model_tools._last_resolved_tool_names with child's toolset.
-    import model_tools as _model_tools
-
-    _parent_tool_names = list(_model_tools._last_resolved_tool_names)
-
     # Capture the ORIGINATING session's wake target BEFORE any child agent is
     # constructed: _build_child_agent() -> AIAgent() -> agent_init calls
     # set_current_session_id(child.session_id), which clobbers the
@@ -2765,53 +2902,49 @@ def delegate_task(
 
     _origin_wake_sid = _current_origin_session_id()
 
-    # Build all child agents on the main thread (thread-safe construction)
-    # Wrapped in try/finally so the global is always restored even if a
-    # child build raises (otherwise _last_resolved_tool_names stays corrupted).
+    # Build all child agents on the main thread (thread-safe construction).
+    # _build_child_preserving_parent_tools saves/restores the parent's
+    # resolved tool names around each construction under a lock, so child
+    # toolset resolution never leaks into the parent (shared with the plugin
+    # subagent-lifecycle API).
     children = []
-    try:
-        for i, t in enumerate(task_list):
-            # Per-task role beats top-level; normalise again so unknown
-            # per-task values warn and degrade to leaf uniformly.
-            effective_role = _normalize_role(t.get("role") or top_role)
-            child = _build_child_agent(
-                task_index=i,
-                goal=t["goal"],
-                context=t.get("context"),
-                # Subagents always inherit the parent's toolsets; the model
-                # cannot choose or narrow them (no model-facing toolsets arg).
-                toolsets=None,
-                model=creds["model"],
-                max_iterations=effective_max_iter,
-                task_count=n_tasks,
-                parent_agent=parent_agent,
-                override_provider=creds["provider"],
-                override_base_url=creds["base_url"],
-                override_api_key=creds["api_key"],
-                override_api_mode=creds["api_mode"],
-                override_request_overrides=creds.get("request_overrides"),
-                override_max_tokens=creds.get("max_output_tokens"),
-                override_acp_command=creds.get("command"),
-                override_acp_args=creds.get("args"),
-                role=effective_role,
+    for i, t in enumerate(task_list):
+        # Per-task role beats top-level; normalise again so unknown
+        # per-task values warn and degrade to leaf uniformly.
+        effective_role = _normalize_role(t.get("role") or top_role)
+        child = _build_child_preserving_parent_tools(
+            task_index=i,
+            goal=t["goal"],
+            context=t.get("context"),
+            # Subagents always inherit the parent's toolsets; the model
+            # cannot choose or narrow them (no model-facing toolsets arg).
+            toolsets=None,
+            model=creds["model"],
+            max_iterations=effective_max_iter,
+            task_count=n_tasks,
+            parent_agent=parent_agent,
+            override_provider=creds["provider"],
+            override_base_url=creds["base_url"],
+            override_api_key=creds["api_key"],
+            override_api_mode=creds["api_mode"],
+            override_request_overrides=creds.get("request_overrides"),
+            override_max_tokens=creds.get("max_output_tokens"),
+            override_acp_command=creds.get("command"),
+            override_acp_args=creds.get("args"),
+            role=effective_role,
+        )
+        # Tee the child's progress events into its live transcript log.
+        # wrap_progress_callback preserves the inner callback contract
+        # (including the _flush attribute) and never lets writer failures
+        # reach the agent loop. When no parent display exists the inner
+        # callback is None and the wrapper still records events.
+        _writer = live_writers[i] if i < len(live_writers) else None
+        if _writer is not None:
+            child.tool_progress_callback = wrap_progress_callback(
+                getattr(child, "tool_progress_callback", None), _writer
             )
-            # Override with correct parent tool names (before child construction mutated global)
-            child._delegate_saved_tool_names = _parent_tool_names
-            # Tee the child's progress events into its live transcript log.
-            # wrap_progress_callback preserves the inner callback contract
-            # (including the _flush attribute) and never lets writer failures
-            # reach the agent loop. When no parent display exists the inner
-            # callback is None and the wrapper still records events.
-            _writer = live_writers[i] if i < len(live_writers) else None
-            if _writer is not None:
-                child.tool_progress_callback = wrap_progress_callback(
-                    getattr(child, "tool_progress_callback", None), _writer
-                )
-                child._live_transcript_path = str(_writer.path)
-            children.append((i, t, child))
-    finally:
-        # Authoritative restore: reset global to parent's tool names after all children built
-        _model_tools._last_resolved_tool_names = _parent_tool_names
+            child._live_transcript_path = str(_writer.path)
+        children.append((i, t, child))
 
     def _execute_and_aggregate() -> dict:
         """Run all built children (1 or N), join on them, aggregate results,
@@ -2955,104 +3088,7 @@ def delegate_task(
         # headroom (split across the batch) before they enter the parent's
         # conversation. Full text is spilled to disk so nothing is lost.
         # Covers both the single-task and batch paths. See PR #9126.
-        _apply_summary_budget(results, parent_agent)
-
-        # Notify parent's memory provider of delegation outcomes
-        if (
-            parent_agent
-            and hasattr(parent_agent, "_memory_manager")
-            and parent_agent._memory_manager
-        ):
-            for entry in results:
-                try:
-                    _task_goal = (
-                        task_list[entry["task_index"]]["goal"]
-                        if entry["task_index"] < len(task_list)
-                        else ""
-                    )
-                    parent_agent._memory_manager.on_delegation(
-                        task=_task_goal,
-                        result=entry.get("summary", "") or "",
-                        child_session_id=(
-                            getattr(children[entry["task_index"]][2], "session_id", "")
-                            if entry["task_index"] < len(children)
-                            else ""
-                        ),
-                    )
-                except Exception:
-                    pass
-
-        # Fire subagent_stop hooks once per child, serialised on the parent thread.
-        # This keeps Python-plugin and shell-hook callbacks off of the worker threads
-        # that ran the children, so hook authors don't need to reason about
-        # concurrent invocation.  Role was captured into the entry dict in
-        # _run_single_child (or the fabricated-entry branches above) before the
-        # child was closed.
-        _parent_session_id = getattr(parent_agent, "session_id", None)
-        try:
-            from hermes_cli.plugins import invoke_hook as _invoke_hook
-        except Exception:
-            _invoke_hook = None
-        # Aggregate child spend here so the parent's footer/UI reflect the true
-        # cost of a subagent-heavy turn.  Port of Kilo-Org/kilocode#9448.  Each
-        # child's cost was captured in _run_single_child before its AIAgent was
-        # closed; we fold them into the parent in one pass alongside the
-        # subagent_stop hook loop so we don't walk `results` twice.
-        _children_cost_total = 0.0
-        for entry in results:
-            child_role = entry.pop("_child_role", None)
-            child_cost = entry.pop("_child_cost_usd", 0.0)
-            try:
-                if child_cost:
-                    _children_cost_total += float(child_cost)
-            except (TypeError, ValueError):
-                pass
-            if _invoke_hook is None:
-                continue
-            try:
-                _child_index = entry.get("task_index", -1)
-                _child_agent = (
-                    children[_child_index][2]
-                    if isinstance(_child_index, int) and 0 <= _child_index < len(children)
-                    else None
-                )
-                _invoke_hook(
-                    "subagent_stop",
-                    parent_session_id=_parent_session_id,
-                    parent_turn_id=getattr(parent_agent, "_current_turn_id", "") or "",
-                    child_session_id=getattr(_child_agent, "session_id", None),
-                    child_role=child_role,
-                    child_summary=entry.get("summary"),
-                    child_status=entry.get("status"),
-                    tool_call_history=_subagent_stop_tool_call_history(
-                        entry.get("tool_trace")
-                    ),
-                    duration_ms=int((entry.get("duration_seconds") or 0) * 1000),
-                )
-            except Exception:
-                logger.debug("subagent_stop hook invocation failed", exc_info=True)
-
-        # Fold the aggregated child cost into the parent's session total.  This is
-        # additive — each delegate_task call contributes its own children — so
-        # nested orchestrator→worker trees roll up naturally: each layer's own
-        # delegate_task() folds its direct children in, and when the orchestrator
-        # itself finishes, its parent folds the orchestrator's now-inflated total
-        # on top.  Degrades silently if the parent lacks the counter (older test
-        # fixtures, etc.).
-        if _children_cost_total > 0.0:
-            try:
-                current = float(getattr(parent_agent, "session_estimated_cost_usd", 0.0) or 0.0)
-                parent_agent.session_estimated_cost_usd = current + _children_cost_total
-                # Upgrade the cost_source so the UI doesn't label a partially-real
-                # total as "none" when the parent itself hadn't billed any calls
-                # yet (rare but possible when the parent's only action this turn
-                # was delegate_task).
-                if getattr(parent_agent, "session_cost_source", "none") in {None, "", "none"}:
-                    parent_agent.session_cost_source = "subagent"
-                if getattr(parent_agent, "session_cost_status", "unknown") in {None, "", "unknown"}:
-                    parent_agent.session_cost_status = "estimated"
-            except Exception:
-                logger.debug("Subagent cost rollup failed", exc_info=True)
+        _finalize_child_results(results, task_list, children, parent_agent)
 
         total_duration = round(time.monotonic() - overall_start, 2)
 

website/docs/developer-guide/subagent-lifecycle-api.md

@@ -7,11 +7,15 @@ sidebar_label: Subagent lifecycle API
 
 Plugins can launch and supervise fresh Hermes child sessions without importing
 `tools.delegate_tool`, gateway internals, TUI state, or `AIAgent` fields.
+The service resolves its parent from the current agent turn, so it works in
+CLI, gateway, non-interactive, and kanban-worker sessions. Launching outside an
+active agent turn fails closed with `No active Hermes parent session`.
 
 ```python
 from agent.subagent_lifecycle import SubagentLaunchRequest
 
-def register(ctx):
+def launch_review(ctx):
+    # Call from a plugin tool or hook while an agent turn is active.
     service = ctx.subagent_lifecycle
     handle = service.launch(SubagentLaunchRequest(
         goal="Review this change for regressions.",
@@ -39,7 +43,10 @@ claims completion until `wait` or `result` observes a terminal state. Terminal
 results are immutable, idempotent, bounded to 32k characters, omit transcripts
 and hidden reasoning, and include a stable result hash.
 
-This API is lifecycle-managed asynchronous execution. It does not change the
+This API is lifecycle-managed asynchronous execution. Child construction and
+completion use the same host-owned path as `delegate_task`, including parent
+tool-resolution restoration, memory notification, serialized `subagent_stop`
+hooks, resource cleanup, and child-cost rollup. It does not change the
 synchronous `delegate_task` tool, batch delegation, or its gateway/TUI display.
 The initial implementation retains metadata and terminal results in-process for
 one hour.