diff --git a/agent/subagent_lifecycle.py b/agent/subagent_lifecycle.py new file mode 100644 index 0000000000000..63603675bc74c --- /dev/null +++ b/agent/subagent_lifecycle.py @@ -0,0 +1,528 @@ +"""Public, plugin-safe lifecycle API for delegated Hermes subagents. + +This module deliberately exposes immutable contracts, not ``AIAgent`` objects. +It is the supported boundary for plugins that need to supervise fresh child +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 + + +PUBLIC_CONTRACT_VERSION = 1 +_MAX_GOAL_CHARS = 16_000 +_MAX_CONTEXT_CHARS = 32_000 +_MAX_METADATA_BYTES = 8_192 +_MAX_RESULT_CHARS = 32_000 +_TERMINAL_RETENTION_SECONDS = 3_600 + + +class SubagentLifecycleError(ValueError): + """A request cannot be safely accepted by the public lifecycle API.""" + + +class SubagentState(str, enum.Enum): + PENDING = "PENDING" + STARTING = "STARTING" + RUNNING = "RUNNING" + SUCCEEDED = "SUCCEEDED" + FAILED = "FAILED" + INTERRUPTED = "INTERRUPTED" + CANCEL_REQUESTED = "CANCEL_REQUESTED" + CANCELLED = "CANCELLED" + UNKNOWN = "UNKNOWN" + + +@dataclasses.dataclass(frozen=True) +class SubagentLaunchRequest: + goal: str + context: Optional[str] = None + role: str = "leaf" + model: Optional[str] = None + allowed_toolsets: Optional[tuple[str, ...]] = None + blocked_tools: tuple[str, ...] = () + working_directory: Optional[str] = None + parent_session_id: Optional[str] = None + correlation_id: Optional[str] = None + metadata: Mapping[str, Any] = dataclasses.field(default_factory=dict) + timeout_seconds: Optional[float] = None + + +@dataclasses.dataclass(frozen=True) +class SubagentHandle: + contract_version: int + subagent_id: str + parent_session_id: Optional[str] + correlation_id: Optional[str] + created_at: float + provider: Optional[str] + model: Optional[str] + role: str + depth: int + capability: str + + def to_dict(self) -> dict[str, Any]: + return dataclasses.asdict(self) + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> "SubagentHandle": + try: + return cls(**dict(value)) + except (TypeError, ValueError) as exc: + raise SubagentLifecycleError("Malformed subagent handle.") from exc + + +@dataclasses.dataclass(frozen=True) +class SubagentStatus: + handle: SubagentHandle + state: SubagentState + updated_at: float + diagnostic: Optional[str] = None + + +@dataclasses.dataclass(frozen=True) +class SubagentTerminalState: + handle: SubagentHandle + state: SubagentState + completed: bool + timed_out: bool = False + diagnostic: Optional[str] = None + + +@dataclasses.dataclass(frozen=True) +class SubagentCancelResult: + accepted: bool + already_terminal: bool = False + unknown_handle: bool = False + unsupported: bool = False + state: SubagentState = SubagentState.UNKNOWN + + +@dataclasses.dataclass(frozen=True) +class SubagentResult: + handle: SubagentHandle + terminal_state: SubagentState + ready: bool + summary: Optional[str] = None + structured_payload: Optional[Mapping[str, Any]] = None + started_at: Optional[float] = None + completed_at: Optional[float] = None + error_classification: Optional[str] = None + error_message: Optional[str] = None + usage_metadata: Mapping[str, Any] = dataclasses.field(default_factory=dict) + tool_execution_summary: Mapping[str, Any] = dataclasses.field(default_factory=dict) + result_hash: Optional[str] = None + + +@dataclasses.dataclass(frozen=True) +class SubagentReconnectResult: + connected: bool + state: SubagentState + diagnostic: Optional[str] = None + + +@dataclasses.dataclass +class _Record: + handle: SubagentHandle + state: SubagentState + updated_at: float + agent: Any = None + future: Optional[Future] = None + started_at: Optional[float] = None + completed_at: Optional[float] = None + result: Optional[SubagentResult] = None + + +class _Registry: + """Thread-safe terminal-retention registry; never returns live records.""" + + def __init__(self) -> None: + self.lock = threading.RLock() + self.records: dict[str, _Record] = {} + self.correlations: dict[tuple[Optional[str], str], str] = {} + + +_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: + """Stable public service returned by :attr:`PluginContext.subagent_lifecycle`. + + Running children are in-process only. Completed results remain available + until process exit; ``reconnect`` accurately reports that a serialized + handle cannot reconnect after a restart instead of launching work again. + """ + + def __init__(self, parent_agent_resolver: Callable[[], Any]) -> None: + self._parent_agent_resolver = parent_agent_resolver + + def launch(self, request: SubagentLaunchRequest) -> SubagentHandle: + parent = self._parent_agent_resolver() + if parent is None: + raise SubagentLifecycleError( + "No active Hermes parent session is available." + ) + self._validate_request(request, parent) + parent_session_id = str(getattr(parent, "session_id", "") or "") or None + if request.parent_session_id and request.parent_session_id != parent_session_id: + raise SubagentLifecycleError( + "parent_session_id does not match the active session." + ) + correlation_key = (parent_session_id, request.correlation_id or "") + with _REGISTRY.lock: + self._cleanup_locked() + if request.correlation_id and correlation_key in _REGISTRY.correlations: + raise SubagentLifecycleError( + "Duplicate correlation_id for this parent session." + ) + + # 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_preserving_parent_tools, + DEFAULT_MAX_ITERATIONS, + ) + + child = _build_child_preserving_parent_tools( + task_index=0, + goal=request.goal, + context=request.context, + toolsets=list(request.allowed_toolsets) + if request.allowed_toolsets + else None, + model=request.model, + max_iterations=DEFAULT_MAX_ITERATIONS, + task_count=1, + parent_agent=parent, + role=request.role, + ) + subagent_id = str(getattr(child, "_subagent_id", "") or "") + if not subagent_id: + raise SubagentLifecycleError("Hermes failed to assign a child identity.") + created = time.time() + handle = SubagentHandle( + PUBLIC_CONTRACT_VERSION, + subagent_id, + parent_session_id, + request.correlation_id, + created, + getattr(child, "provider", None), + getattr(child, "model", None), + getattr(child, "_delegate_role", request.role), + int(getattr(child, "_delegate_depth", 1) or 1), + self._capability(subagent_id, parent_session_id, created), + ) + record = _Record(handle, SubagentState.PENDING, created, agent=child) + with _REGISTRY.lock: + _REGISTRY.records[subagent_id] = record + if request.correlation_id: + _REGISTRY.correlations[correlation_key] = subagent_id + record.future = _EXECUTOR.submit(self._run, record, request.goal, parent) + return handle + + def status(self, handle: SubagentHandle) -> SubagentStatus: + record = self._record(handle) + if record is None: + return SubagentStatus( + handle, SubagentState.UNKNOWN, time.time(), "UNKNOWN_HANDLE" + ) + with _REGISTRY.lock: + return SubagentStatus(record.handle, record.state, record.updated_at) + + def wait( + self, handle: SubagentHandle, *, timeout_seconds: Optional[float] = None + ) -> SubagentTerminalState: + record = self._record(handle) + if record is None: + return SubagentTerminalState( + handle, SubagentState.UNKNOWN, True, diagnostic="UNKNOWN_HANDLE" + ) + future = record.future + if future is not None: + try: + future.result(timeout=timeout_seconds) + except TimeoutError: + return SubagentTerminalState(record.handle, record.state, False, True) + except Exception: + pass + with _REGISTRY.lock: + return SubagentTerminalState( + record.handle, record.state, record.result is not None + ) + + def cancel(self, handle: SubagentHandle, *, reason: str) -> SubagentCancelResult: + record = self._record(handle) + if record is None: + return SubagentCancelResult(False, unknown_handle=True) + with _REGISTRY.lock: + if record.result is not None: + return SubagentCancelResult( + False, already_terminal=True, state=record.state + ) + agent = record.agent + record.state = SubagentState.CANCEL_REQUESTED + record.updated_at = time.time() + if agent is None or not hasattr(agent, "interrupt"): + return SubagentCancelResult( + False, unsupported=True, state=SubagentState.CANCEL_REQUESTED + ) + try: + agent.interrupt(f"Lifecycle cancellation requested: {reason[:500]}") + except Exception: + return SubagentCancelResult( + False, unsupported=True, state=SubagentState.CANCEL_REQUESTED + ) + return SubagentCancelResult(True, state=SubagentState.CANCEL_REQUESTED) + + def result(self, handle: SubagentHandle) -> SubagentResult: + record = self._record(handle) + if record is None: + return SubagentResult( + handle, + SubagentState.UNKNOWN, + False, + error_classification="UNKNOWN_HANDLE", + ) + with _REGISTRY.lock: + if record.result is not None: + return record.result + return SubagentResult( + record.handle, record.state, False, error_classification="NOT_READY" + ) + + def reconnect(self, handle: SubagentHandle) -> SubagentReconnectResult: + record = self._record(handle) + if record is None: + return SubagentReconnectResult( + False, SubagentState.UNKNOWN, "RECONNECT_UNAVAILABLE" + ) + with _REGISTRY.lock: + return SubagentReconnectResult(True, record.state) + + 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( + handle.subagent_id, handle.parent_session_id, handle.created_at + ), + ): + return None + parent = self._parent_agent_resolver() + active_parent_id = str(getattr(parent, "session_id", "") or "") or None + if active_parent_id != handle.parent_session_id: + return None + with _REGISTRY.lock: + return _REGISTRY.records.get(handle.subagent_id) + + @staticmethod + def _cleanup_locked() -> None: + """Retain terminal snapshots for a bounded period, never live work.""" + cutoff = time.time() - _TERMINAL_RETENTION_SECONDS + expired = [ + subagent_id + for subagent_id, record in _REGISTRY.records.items() + if record.result is not None + and record.completed_at is not None + and record.completed_at < cutoff + ] + for subagent_id in expired: + record = _REGISTRY.records.pop(subagent_id) + if record.handle.correlation_id: + _REGISTRY.correlations.pop( + (record.handle.parent_session_id, record.handle.correlation_id), + None, + ) + + def _run(self, record: _Record, goal: str, parent: Any) -> None: + with _REGISTRY.lock: + 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_child_lifecycle + + raw = _run_child_lifecycle(0, goal, record.agent, parent) + status = ( + str(raw.get("status", "error")) if isinstance(raw, dict) else "error" + ) + if status == "completed": + state = SubagentState.SUCCEEDED + elif status == "interrupted": + state = ( + SubagentState.CANCELLED + if record.state == SubagentState.CANCEL_REQUESTED + else SubagentState.INTERRUPTED + ) + else: + state = SubagentState.FAILED + summary = raw.get("summary") if isinstance(raw, dict) else None + summary = str(summary)[:_MAX_RESULT_CHARS] if summary is not None else None + error = raw.get("error") if isinstance(raw, dict) else None + result = SubagentResult( + record.handle, + state, + True, + summary=summary, + completed_at=time.time(), + started_at=record.started_at, + error_classification=None + if state == SubagentState.SUCCEEDED + else status.upper(), + error_message=str(error)[:_MAX_RESULT_CHARS] if error else None, + usage_metadata={"api_calls": raw.get("api_calls", 0)} + if isinstance(raw, dict) + else {}, + tool_execution_summary={ + "duration_seconds": raw.get("duration_seconds", 0) + } + if isinstance(raw, dict) + else {}, + ) + except Exception as exc: + result = SubagentResult( + record.handle, + SubagentState.FAILED, + True, + started_at=record.started_at, + completed_at=time.time(), + error_classification=type(exc).__name__, + error_message=str(exc)[:_MAX_RESULT_CHARS], + ) + payload = dataclasses.asdict(result) + payload.pop("result_hash", None) + result = dataclasses.replace( + result, + result_hash=hashlib.sha256( + json.dumps(payload, sort_keys=True, default=str).encode() + ).hexdigest(), + ) + with _REGISTRY.lock: + record.agent = None + record.result = result + record.state = result.terminal_state + record.completed_at = result.completed_at + record.updated_at = result.completed_at or time.time() + + @staticmethod + def _capability( + subagent_id: str, parent_session_id: Optional[str], created_at: float + ) -> str: + value = f"{subagent_id}|{parent_session_id or ''}|{created_at:.6f}".encode() + return hmac.new(_SECRET, value, hashlib.sha256).hexdigest() + + @staticmethod + def _validate_request(request: SubagentLaunchRequest, parent: Any) -> None: + if ( + not isinstance(request, SubagentLaunchRequest) + or not isinstance(request.goal, str) + or not request.goal.strip() + or len(request.goal) > _MAX_GOAL_CHARS + ): + raise SubagentLifecycleError( + "goal must be a non-empty string of at most 16000 characters." + ) + if request.context is not None and ( + not isinstance(request.context, str) + or len(request.context) > _MAX_CONTEXT_CHARS + ): + raise SubagentLifecycleError( + "context must be a string of at most 32000 characters." + ) + if request.role not in {"leaf", "orchestrator"}: + raise SubagentLifecycleError("role must be 'leaf' or 'orchestrator'.") + if request.timeout_seconds is not None: + raise SubagentLifecycleError( + "Per-launch timeout is not supported; configure delegation timeout explicitly." + ) + if request.working_directory is not None: + raise SubagentLifecycleError( + "working_directory is not supported because Hermes delegates use isolated task environments." + ) + if request.blocked_tools: + raise SubagentLifecycleError( + "Per-tool blocking is not supported; use allowed_toolsets. Hermes always blocks unsafe child tools." + ) + try: + metadata_bytes = len( + json.dumps(dict(request.metadata), sort_keys=True).encode() + ) + except (TypeError, ValueError) as exc: + raise SubagentLifecycleError("metadata must be JSON-serializable.") from exc + if metadata_bytes > _MAX_METADATA_BYTES: + raise SubagentLifecycleError("metadata exceeds 8192 bytes.") + if request.allowed_toolsets: + from toolsets import TOOLSETS + + unknown = set(request.allowed_toolsets) - set(TOOLSETS) + if unknown: + raise SubagentLifecycleError( + f"Unknown toolsets: {', '.join(sorted(unknown))}." + ) + enabled = getattr(parent, "enabled_toolsets", None) + if enabled is not None and not set(request.allowed_toolsets).issubset( + set(enabled) + ): + raise SubagentLifecycleError( + "Requested toolsets would broaden parent permissions." + ) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 6ca393fca53c1..fedbcb981fe97 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -344,6 +344,7 @@ def __init__(self, manifest: PluginManifest, manager: "PluginManager"): self._manager = manager # Lazy-built host-owned LLM facade — see ctx.llm property below. self._llm: Any = None + self._subagent_lifecycle: Any = None # -- host-owned LLM access ---------------------------------------------- @@ -364,6 +365,24 @@ def llm(self) -> Any: self._llm = PluginLlm(plugin_id=plugin_id) return self._llm + @property + def subagent_lifecycle(self) -> Any: + """Return the public, plugin-safe subagent lifecycle service. + + The service only resolves the active host-owned parent agent when a + child is launched. Plugins receive serializable handles and immutable + snapshots; they never receive a live agent or a private registry. + """ + if self._subagent_lifecycle is None: + from agent.subagent_lifecycle import ( + SubagentLifecycleService, + get_active_subagent_parent, + ) + self._subagent_lifecycle = SubagentLifecycleService( + get_active_subagent_parent + ) + return self._subagent_lifecycle + # -- profile awareness -------------------------------------------------- @property diff --git a/run_agent.py b/run_agent.py index 8a92981e1dd2d..71589fa7eeebc 100644 --- a/run_agent.py +++ b/run_agent.py @@ -6193,6 +6193,8 @@ def run_conversation( 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 @@ -6212,7 +6214,7 @@ def run_conversation( # 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, diff --git a/tests/agent/test_subagent_lifecycle.py b/tests/agent/test_subagent_lifecycle.py new file mode 100644 index 0000000000000..faa3ddabbf558 --- /dev/null +++ b/tests/agent/test_subagent_lifecycle.py @@ -0,0 +1,252 @@ +"""Contract tests for the public plugin subagent lifecycle API.""" + +import time +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from agent.subagent_lifecycle import ( + SubagentLaunchRequest, + SubagentLifecycleError, + SubagentLifecycleService, + SubagentState, + bind_subagent_parent, + get_active_subagent_parent, +) + + +class FakeChild: + def __init__(self, ident="sa-test"): + self._subagent_id = ident + self._delegate_role = "leaf" + self._delegate_depth = 1 + self.provider = "test" + self.model = "test-model" + self.interrupted = False + + def interrupt(self, _reason): + self.interrupted = True + + +@pytest.fixture +def lifecycle(monkeypatch): + parent = SimpleNamespace(session_id="parent-1", enabled_toolsets=["file"]) + counter = iter(range(1000)) + + def build(**_kwargs): + return FakeChild(f"sa-{next(counter)}") + + def run(_index, _goal, child, _parent): + for _ in range(20): + if child.interrupted: + return { + "status": "interrupted", + "summary": None, + "api_calls": 0, + "duration_seconds": 0, + } + time.sleep(0.002) + return { + "status": "completed", + "summary": "safe summary", + "api_calls": 1, + "duration_seconds": 0.01, + } + + monkeypatch.setattr("tools.delegate_tool._build_child_agent", build) + monkeypatch.setattr("tools.delegate_tool._run_single_child", run) + return SubagentLifecycleService(lambda: parent) + + +def test_launch_wait_result_and_handle_round_trip(lifecycle): + handle = lifecycle.launch( + SubagentLaunchRequest(goal="x", allowed_toolsets=("file",)) + ) + assert handle.from_dict(handle.to_dict()) == handle + assert lifecycle.wait(handle, timeout_seconds=1).state is SubagentState.SUCCEEDED + first = lifecycle.result(handle) + assert first.ready and first.summary == "safe summary" and first.result_hash + assert lifecycle.result(handle) == first + + +def test_duplicate_correlation_and_permission_validation(lifecycle): + 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"): + lifecycle.launch( + SubagentLaunchRequest(goal="x", allowed_toolsets=("terminal",)) + ) + 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): + handle = lifecycle.launch(SubagentLaunchRequest(goal="x")) + assert lifecycle.cancel(handle, reason="test").accepted + terminal = lifecycle.wait(handle, timeout_seconds=1) + assert terminal.state is SubagentState.CANCELLED + forged = handle.__class__(**{**handle.to_dict(), "capability": "forged"}) + assert lifecycle.status(forged).state is SubagentState.UNKNOWN + assert lifecycle.result(forged).error_classification == "UNKNOWN_HANDLE" + other_parent = SimpleNamespace(session_id="different-parent") + other_service = SubagentLifecycleService(lambda: other_parent) + assert other_service.status(handle).state is SubagentState.UNKNOWN + + +def test_simultaneous_launches_are_distinct_and_reconnect_is_in_process(lifecycle): + 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 diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 12f94a180a8ef..d2ed4a7a6616f 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2400,6 +2400,147 @@ def _run_with_thread_capture(): 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"), + 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]]: @@ -2555,49 +2696,34 @@ def delegate_task( # Track goal labels for progress display (truncated for readability) task_labels = [t["goal"][:40] for t in task_list] - # 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) - # 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). 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, - ) - # Override with correct parent tool names (before child construction mutated global) - child._delegate_saved_tool_names = _parent_tool_names - 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 + 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, + ) + children.append((i, t, child)) def _execute_and_aggregate() -> dict: """Run all built children (1 or N), join on them, aggregate results, @@ -2741,101 +2867,7 @@ def _execute_and_aggregate() -> dict: # 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"), - 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) diff --git a/website/docs/developer-guide/subagent-lifecycle-api.md b/website/docs/developer-guide/subagent-lifecycle-api.md new file mode 100644 index 0000000000000..7054afcdb7356 --- /dev/null +++ b/website/docs/developer-guide/subagent-lifecycle-api.md @@ -0,0 +1,61 @@ +--- +title: Public Subagent Lifecycle API +sidebar_label: Subagent lifecycle API +--- + +# Public 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 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.", + context="Only inspect the supplied repository.", + role="leaf", + correlation_id="review-42", + allowed_toolsets=("file",), + )) + # Persist handle.to_dict() if desired. + if service.wait(handle, timeout_seconds=2).timed_out: + return handle.to_dict() + return service.result(handle) +``` + +`SubagentHandle` is serializable and carries a versioned, opaque capability. +Pass it back to `status`, `wait`, `cancel`, `result`, or `reconnect`; malformed +or forged handles return `UNKNOWN`/`UNKNOWN_HANDLE` and cannot access a child. + +The stable states are `PENDING`, `STARTING`, `RUNNING`, `SUCCEEDED`, `FAILED`, +`INTERRUPTED`, `CANCEL_REQUESTED`, `CANCELLED`, and `UNKNOWN`. + +`cancel(handle, reason=...)` is cooperative: it asks the child agent to +interrupt at its next safe boundary and returns `CANCEL_REQUESTED`; it never +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. 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. +After a process restart, `reconnect` returns `RECONNECT_UNAVAILABLE` and never +starts a replacement child. Running Python threads also cannot survive process +exit; callers must treat those handles as interrupted by process exit. + +Requests are fail-closed: goal/context/metadata sizes are capped, unknown or +parent-broadening toolsets are rejected, and per-tool blocks, working-directory +overrides, and per-launch timeouts are explicitly rejected until Hermes can +support them without weakening isolation. Use `allowed_toolsets` to narrow a +child; Hermes's existing unsafe-tool block remains enforced.