test_subagent_stop_hook.py
1 """Tests for the subagent_stop hook event. 2 3 Covers wire-up from tools.delegate_tool.delegate_task: 4 * fires once per child in both single-task and batch modes 5 * runs on the parent thread (no re-entrancy for hook authors) 6 * carries child_role when the agent exposes _delegate_role 7 * carries child_role=None when _delegate_role is not set (pre-M3) 8 """ 9 10 from __future__ import annotations 11 12 import json 13 import threading 14 from unittest.mock import MagicMock, patch 15 16 import pytest 17 18 from tools.delegate_tool import delegate_task 19 from hermes_cli import plugins 20 21 22 def _make_parent(depth: int = 0, session_id: str = "parent-1"): 23 parent = MagicMock() 24 parent.base_url = "https://openrouter.ai/api/v1" 25 parent.api_key = "***" 26 parent.provider = "openrouter" 27 parent.api_mode = "chat_completions" 28 parent.model = "anthropic/claude-sonnet-4" 29 parent.platform = "cli" 30 parent.providers_allowed = None 31 parent.providers_ignored = None 32 parent.providers_order = None 33 parent.provider_sort = None 34 parent._session_db = None 35 parent._delegate_depth = depth 36 parent._active_children = [] 37 parent._active_children_lock = threading.Lock() 38 parent._print_fn = None 39 parent.tool_progress_callback = None 40 parent.thinking_callback = None 41 parent._memory_manager = None 42 parent.session_id = session_id 43 return parent 44 45 46 @pytest.fixture(autouse=True) 47 def _fresh_plugin_manager(): 48 """Each test gets a fresh PluginManager so hook callbacks don't 49 leak between tests.""" 50 original = plugins._plugin_manager 51 plugins._plugin_manager = plugins.PluginManager() 52 yield 53 plugins._plugin_manager = original 54 55 56 @pytest.fixture(autouse=True) 57 def _stub_child_builder(monkeypatch): 58 """Replace _build_child_agent with a MagicMock factory so delegate_task 59 never transitively imports run_agent / openai. Keeps the test runnable 60 in environments without heavyweight runtime deps installed.""" 61 def _fake_build_child(task_index, **kwargs): 62 child = MagicMock() 63 child._delegate_saved_tool_names = [] 64 child._credential_pool = None 65 return child 66 67 monkeypatch.setattr( 68 "tools.delegate_tool._build_child_agent", _fake_build_child, 69 ) 70 71 72 def _register_capturing_hook(): 73 captured = [] 74 75 def _cb(**kwargs): 76 kwargs["_thread"] = threading.current_thread() 77 captured.append(kwargs) 78 79 mgr = plugins.get_plugin_manager() 80 mgr._hooks.setdefault("subagent_stop", []).append(_cb) 81 return captured 82 83 84 # ── single-task mode ────────────────────────────────────────────────────── 85 86 87 class TestSingleTask: 88 def test_fires_once(self): 89 captured = _register_capturing_hook() 90 91 with patch("tools.delegate_tool._run_single_child") as mock_run: 92 mock_run.return_value = { 93 "task_index": 0, 94 "status": "completed", 95 "summary": "Done!", 96 "api_calls": 3, 97 "duration_seconds": 5.0, 98 "_child_role": "analyst", 99 } 100 delegate_task(goal="do X", parent_agent=_make_parent()) 101 102 assert len(captured) == 1 103 payload = captured[0] 104 assert payload["child_role"] == "analyst" 105 assert payload["child_status"] == "completed" 106 assert payload["child_summary"] == "Done!" 107 assert payload["duration_ms"] == 5000 108 109 def test_fires_on_parent_thread(self): 110 captured = _register_capturing_hook() 111 main_thread = threading.current_thread() 112 113 with patch("tools.delegate_tool._run_single_child") as mock_run: 114 mock_run.return_value = { 115 "task_index": 0, "status": "completed", 116 "summary": "x", "api_calls": 1, "duration_seconds": 0.1, 117 "_child_role": None, 118 } 119 delegate_task(goal="go", parent_agent=_make_parent()) 120 121 assert captured[0]["_thread"] is main_thread 122 123 def test_payload_includes_parent_session_id(self): 124 captured = _register_capturing_hook() 125 126 with patch("tools.delegate_tool._run_single_child") as mock_run: 127 mock_run.return_value = { 128 "task_index": 0, "status": "completed", 129 "summary": "x", "api_calls": 1, "duration_seconds": 0.1, 130 "_child_role": None, 131 } 132 delegate_task( 133 goal="go", 134 parent_agent=_make_parent(session_id="sess-xyz"), 135 ) 136 137 assert captured[0]["parent_session_id"] == "sess-xyz" 138 139 140 # ── batch mode ──────────────────────────────────────────────────────────── 141 142 143 class TestBatchMode: 144 def test_fires_per_child(self): 145 captured = _register_capturing_hook() 146 147 with patch("tools.delegate_tool._run_single_child") as mock_run: 148 mock_run.side_effect = [ 149 {"task_index": 0, "status": "completed", 150 "summary": "A", "api_calls": 1, "duration_seconds": 1.0, 151 "_child_role": "role-a"}, 152 {"task_index": 1, "status": "completed", 153 "summary": "B", "api_calls": 2, "duration_seconds": 2.0, 154 "_child_role": "role-b"}, 155 {"task_index": 2, "status": "completed", 156 "summary": "C", "api_calls": 3, "duration_seconds": 3.0, 157 "_child_role": "role-c"}, 158 ] 159 delegate_task( 160 tasks=[ 161 {"goal": "A"}, {"goal": "B"}, {"goal": "C"}, 162 ], 163 parent_agent=_make_parent(), 164 ) 165 166 assert len(captured) == 3 167 roles = sorted(c["child_role"] for c in captured) 168 assert roles == ["role-a", "role-b", "role-c"] 169 170 def test_all_fires_on_parent_thread(self): 171 captured = _register_capturing_hook() 172 main_thread = threading.current_thread() 173 174 with patch("tools.delegate_tool._run_single_child") as mock_run: 175 mock_run.side_effect = [ 176 {"task_index": 0, "status": "completed", 177 "summary": "A", "api_calls": 1, "duration_seconds": 1.0, 178 "_child_role": None}, 179 {"task_index": 1, "status": "completed", 180 "summary": "B", "api_calls": 2, "duration_seconds": 2.0, 181 "_child_role": None}, 182 ] 183 delegate_task( 184 tasks=[{"goal": "A"}, {"goal": "B"}], 185 parent_agent=_make_parent(), 186 ) 187 188 for payload in captured: 189 assert payload["_thread"] is main_thread 190 191 192 # ── payload shape ───────────────────────────────────────────────────────── 193 194 195 class TestPayloadShape: 196 def test_role_absent_becomes_none(self): 197 captured = _register_capturing_hook() 198 199 with patch("tools.delegate_tool._run_single_child") as mock_run: 200 mock_run.return_value = { 201 "task_index": 0, "status": "completed", 202 "summary": "x", "api_calls": 1, "duration_seconds": 0.1, 203 # Deliberately omit _child_role — pre-M3 shape. 204 } 205 delegate_task(goal="do X", parent_agent=_make_parent()) 206 207 assert captured[0]["child_role"] is None 208 209 def test_result_does_not_leak_child_role_field(self): 210 """The internal _child_role key must be stripped before the 211 result dict is serialised to JSON.""" 212 _register_capturing_hook() 213 214 with patch("tools.delegate_tool._run_single_child") as mock_run: 215 mock_run.return_value = { 216 "task_index": 0, "status": "completed", 217 "summary": "x", "api_calls": 1, "duration_seconds": 0.1, 218 "_child_role": "leaf", 219 } 220 raw = delegate_task(goal="do X", parent_agent=_make_parent()) 221 222 parsed = json.loads(raw) 223 assert "results" in parsed 224 assert "_child_role" not in parsed["results"][0]