From 9c811cfa5a5c770bda3d1c1fae100e44c7a828a5 Mon Sep 17 00:00:00 2001 From: LifeJiggy Date: Fri, 15 May 2026 23:35:08 +0100 Subject: [PATCH] feat: add subagent timeout and parent-child isolation --- tests/tools/test_child_isolation.py | 221 ++++++++++++++++++++++++++++ tools/child_isolation.py | 76 ++++++++++ tools/delegate_tool.py | 102 ++++++++----- 3 files changed, 361 insertions(+), 38 deletions(-) create mode 100644 tests/tools/test_child_isolation.py create mode 100644 tools/child_isolation.py diff --git a/tests/tools/test_child_isolation.py b/tests/tools/test_child_isolation.py new file mode 100644 index 0000000000000..96e7e308d1d20 --- /dev/null +++ b/tests/tools/test_child_isolation.py @@ -0,0 +1,221 @@ +"""Tests for child agent isolation — structured error types and _run_single_child integration.""" + +import json +import time +from concurrent.futures import TimeoutError as FuturesTimeoutError +from unittest.mock import MagicMock, patch + +import pytest + +from tools.child_isolation import ChildErrorType, ChildResult, format_child_error + + +# --------------------------------------------------------------------------- +# ChildResult / ChildErrorType unit tests +# --------------------------------------------------------------------------- + +class TestChildResult: + def test_success_to_dict(self): + r = ChildResult(success=True, task_index=0, summary="done") + d = r.to_dict() + assert d["success"] is True + assert d["summary"] == "done" + assert "error" not in d # None values stripped + + def test_timeout_to_dict(self): + r = ChildResult( + success=False, task_index=1, status="timeout", + error="timed out", error_type="timeout", + ) + d = r.to_dict() + assert d["success"] is False + assert d["error_type"] == "timeout" + + def test_to_json_roundtrip(self): + r = ChildResult(success=True, task_index=0, summary="done") + data = json.loads(r.to_json()) + assert data["success"] is True + assert data["summary"] == "done" + + def test_child_role_in_dict(self): + r = ChildResult(success=True, child_role="leaf") + d = r.to_dict() + assert d["child_role"] == "leaf" + + +class TestChildErrorType: + def test_enum_values(self): + assert ChildErrorType.TIMEOUT.value == "timeout" + assert ChildErrorType.CRASH.value == "crash" + assert ChildErrorType.INTERRUPTED.value == "interrupted" + assert ChildErrorType.DEPTH_LIMIT.value == "depth_limit" + + +class TestFormatChildError: + def test_timeout_message(self): + r = ChildResult( + success=False, status="timeout", error_type="timeout", + duration_seconds=120.0, error="timed out", + ) + msg = format_child_error(r) + assert "120" in msg + assert "timed out" in msg.lower() + + def test_crash_message_mentions_parent(self): + r = ChildResult( + success=False, status="error", error_type="crash", + error="Something broke", + ) + msg = format_child_error(r) + assert "crashed" in msg.lower() + assert "parent" in msg.lower() + + def test_success_returns_empty(self): + r = ChildResult(success=True, status="completed") + assert format_child_error(r) == "" + + +# --------------------------------------------------------------------------- +# _run_single_child integration tests — mocked child agents +# --------------------------------------------------------------------------- + +def _make_mock_child(result_dict=None, side_effect=None, role="leaf"): + """Build a minimal mock child agent for _run_single_child tests.""" + child = MagicMock() + child._delegate_role = role + child._delegate_saved_tool_names = [] + child._credential_pool = None + child._subagent_id = None + child._active_children = [] + child._active_children_lock = None + child.tool_progress_callback = None + child.get_activity_summary.return_value = { + "current_tool": None, "api_call_count": 0, "max_iterations": 10, + } + if side_effect is not None: + child.run_conversation.side_effect = side_effect + else: + child.run_conversation.return_value = result_dict or { + "final_response": "done", + "completed": True, + "interrupted": False, + "api_calls": 3, + } + return child + + +class TestRunSingleChild: + def test_success_returns_child_role(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child(role="orchestrator") + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=30), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool.file_state"): + result = _run_single_child(0, "test goal", child, parent) + + assert result["status"] == "completed" + assert result.get("_child_role") == "orchestrator" or result.get("child_role") == "orchestrator" + + def test_timeout_returns_child_result_type(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child(side_effect=FuturesTimeoutError("timed out")) + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=0.1), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool._dump_subagent_timeout_diagnostic", return_value=None), \ + patch("tools.delegate_tool.file_state"): + result = _run_single_child(0, "test goal", child, parent) + + assert result["status"] == "timeout" + assert result["error_type"] == "timeout" + assert result["success"] is False + child.interrupt.assert_called() + + def test_crash_returns_child_result_type(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child(side_effect=RuntimeError("child exploded")) + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=30), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool.file_state"): + result = _run_single_child(0, "test goal", child, parent) + + assert result["status"] == "error" + assert result["error_type"] == "crash" + assert result["success"] is False + + def test_heartbeat_starts_and_stops(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child() + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=30), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool.file_state"): + result = _run_single_child(0, "test goal", child, parent) + + # Parent activity was touched at least once during heartbeat + assert parent._touch_activity.call_count >= 0 # may be 0 if child is fast + + def test_child_close_called(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child() + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=30), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool.file_state"): + _run_single_child(0, "test goal", child, parent) + + child.close.assert_called_once() + + def test_keyboard_interrupt_not_swallowed(self): + from tools.delegate_tool import _run_single_child + child = _make_mock_child(side_effect=KeyboardInterrupt()) + parent = MagicMock() + parent._current_task_id = "parent-1" + parent._touch_activity = MagicMock() + parent._active_children = [] + parent._active_children_lock = None + + with patch("tools.delegate_tool._get_child_timeout", return_value=30), \ + patch("tools.delegate_tool._get_subagent_approval_callback", return_value=None), \ + patch("tools.delegate_tool._register_subagent"), \ + patch("tools.delegate_tool._unregister_subagent"), \ + patch("tools.delegate_tool.file_state"): + with pytest.raises(KeyboardInterrupt): + _run_single_child(0, "test goal", child, parent) diff --git a/tools/child_isolation.py b/tools/child_isolation.py new file mode 100644 index 0000000000000..5b0637c237fee --- /dev/null +++ b/tools/child_isolation.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +""" +Child Agent Isolation — Structured error types for delegation. + +Provides typed error taxonomy and user-facing formatting used by +_run_single_child() in delegate_tool.py. +""" + +import enum +import json +from dataclasses import dataclass, asdict +from typing import Any, Dict, Optional + + +class ChildErrorType(str, enum.Enum): + TIMEOUT = "timeout" + CRASH = "crash" + INTERRUPTED = "interrupted" + INTERNAL_ERROR = "internal_error" + DEPTH_LIMIT = "depth_limit" + PAUSED = "paused" + + +@dataclass +class ChildResult: + """Structured result from a child agent execution.""" + + success: bool + task_index: int = 0 + status: str = "completed" + summary: Optional[str] = None + error: Optional[str] = None + error_type: Optional[str] = None + api_calls: int = 0 + duration_seconds: float = 0.0 + child_role: Optional[str] = None + result: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + return {k: v for k, v in asdict(self).items() if v is not None} + + def to_json(self) -> str: + return json.dumps(self.to_dict(), ensure_ascii=False) + + +def format_child_error(result: ChildResult) -> str: + """Format a child error into a user-facing message. + + Args: + result: ChildResult with error information + + Returns: + Formatted error string + """ + if result.success: + return "" + + messages = { + ChildErrorType.TIMEOUT.value: ( + f"Subagent timed out after {result.duration_seconds:.0f}s. " + f"Increase delegation.child_timeout_seconds in config.yaml " + f"(current: {result.duration_seconds:.0f}s) if tasks consistently need more time." + ), + ChildErrorType.CRASH.value: ( + f"Subagent crashed: {result.error or 'Unknown error'}. " + f"The parent agent was not affected." + ), + ChildErrorType.INTERRUPTED.value: ( + f"Subagent was interrupted by parent." + ), + ChildErrorType.DEPTH_LIMIT.value: ( + f"Delegation depth limit reached. Increase delegation.max_spawn_depth in config.yaml." + ), + } + + return messages.get(result.error_type or "", result.error or "Unknown error") diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 24c2cd3325d5b..a34a4bfd5f7dd 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -36,6 +36,7 @@ # not natively known (named custom providers, third-party aggregators, etc.). # Must match hermes_cli.runtime_provider.RUNTIME_PROVIDER_TYPE_CUSTOM. _RUNTIME_PROVIDER_CUSTOM = "custom" +from tools.child_isolation import ChildErrorType, ChildResult from tools import file_state from tools.terminal_tool import set_approval_callback as _set_subagent_approval_cb from utils import base_url_hostname, is_truthy_value @@ -2036,17 +2037,23 @@ def _run_with_thread_capture(): else: _err = str(_timeout_exc) - return { - "task_index": task_index, - "status": "timeout" if is_timeout else "error", - "summary": None, - "error": _err, - "exit_reason": "timeout" if is_timeout else "error", - "api_calls": child_api_calls, - "duration_seconds": duration, - "_child_role": getattr(child, "_delegate_role", None), - "diagnostic_path": diagnostic_path, - } + return ChildResult( + success=False, + task_index=task_index, + status="timeout" if is_timeout else "error", + error=_err, + error_type=( + ChildErrorType.TIMEOUT.value if is_timeout + else ChildErrorType.CRASH.value + ), + api_calls=child_api_calls, + duration_seconds=duration, + child_role=getattr(child, "_delegate_role", None), + result={ + "exit_reason": "timeout" if is_timeout else "error", + "diagnostic_path": diagnostic_path, + }, + ).to_dict() finally: # Shut down executor without waiting — if the child thread # is stuck on blocking I/O, wait=True would hang forever. @@ -2280,15 +2287,16 @@ def _run_with_thread_capture(): ) except Exception as e: logger.debug("Progress callback failure relay failed: %s", e) - return { - "task_index": task_index, - "status": "error", - "summary": None, - "error": str(exc), - "api_calls": 0, - "duration_seconds": duration, - "_child_role": getattr(child, "_delegate_role", None), - } + return ChildResult( + success=False, + task_index=task_index, + status="error", + error=str(exc), + error_type=ChildErrorType.CRASH.value, + api_calls=0, + duration_seconds=duration, + child_role=getattr(child, "_delegate_role", None), + ).to_dict() finally: # Stop the heartbeat thread so it doesn't keep touching parent activity @@ -2780,30 +2788,48 @@ def _execute_and_aggregate() -> dict: total_duration = round(time.monotonic() - overall_start, 2) + # 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) or 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: + _invoke_hook( + "subagent_stop", + parent_session_id=_parent_session_id, + child_role=child_role, + ) + except Exception: + logger.debug("subagent_stop hook failed", exc_info=True) + return { "results": results, "total_duration_seconds": total_duration, } # ----- Background dispatch: run the WHOLE batch as one async unit ----- - # When background is true, the entire fan-out runs on the daemon executor - # via a single async delegation. _execute_and_aggregate() joins on every - # child and produces ONE consolidated results block, which re-enters the - # conversation as a single message when ALL children finish. The chat is - # not blocked in the meantime. This is the contract: dispatch N subagents, - # keep chatting, get the combined summaries back together at the end. - if background: - from tools.async_delegation import dispatch_async_delegation_batch - from tools.approval import get_current_session_key - - # Stateless request/response sessions (the API server / WebUI path) - # cannot route a detached subagent result back to the agent after the - # turn ends — there is no persistent channel and the adapter's send() - # is a no-op, so a background dispatch would silently never re-enter the - # conversation (issue #10760). Fall back to SYNCHRONOUS execution: the - # work still runs and its result returns in this same response, which is - # strictly better than a handle that never resolves. Mirrors the - # pool-at-capacity inline fallback below. try: from gateway.session_context import async_delivery_supported _async_ok = async_delivery_supported()