test_plugins.py
1 """Tests for the Hermes plugin system (hermes_cli.plugins).""" 2 3 import logging 4 import os 5 import sys 6 import types 7 from pathlib import Path 8 from unittest.mock import MagicMock, patch 9 10 import pytest 11 import yaml 12 13 from hermes_cli.plugins import ( 14 ENTRY_POINTS_GROUP, 15 VALID_HOOKS, 16 LoadedPlugin, 17 PluginContext, 18 PluginManager, 19 PluginManifest, 20 get_plugin_manager, 21 get_plugin_command_handler, 22 get_plugin_commands, 23 get_pre_tool_call_block_message, 24 resolve_plugin_command_result, 25 discover_plugins, 26 invoke_hook, 27 ) 28 29 30 # ── Helpers ──────────────────────────────────────────────────────────────── 31 32 33 def _make_plugin_dir(base: Path, name: str, *, register_body: str = "pass", 34 manifest_extra: dict | None = None, 35 auto_enable: bool = True) -> Path: 36 """Create a minimal plugin directory with plugin.yaml + __init__.py. 37 38 If *auto_enable* is True (default), also write the plugin's name into 39 ``<hermes_home>/config.yaml`` under ``plugins.enabled``. Plugins are 40 opt-in by default, so tests that expect the plugin to actually load 41 need this. Pass ``auto_enable=False`` for tests that exercise the 42 unenabled path. 43 44 *base* is expected to be ``<hermes_home>/plugins/``; we derive 45 ``<hermes_home>`` from it by walking one level up. 46 """ 47 plugin_dir = base / name 48 plugin_dir.mkdir(parents=True, exist_ok=True) 49 50 manifest = {"name": name, "version": "0.1.0", "description": f"Test plugin {name}"} 51 if manifest_extra: 52 manifest.update(manifest_extra) 53 54 (plugin_dir / "plugin.yaml").write_text(yaml.dump(manifest)) 55 (plugin_dir / "__init__.py").write_text( 56 f"def register(ctx):\n {register_body}\n" 57 ) 58 59 if auto_enable: 60 # Write/merge plugins.enabled in <HERMES_HOME>/config.yaml. 61 # Config is always read from HERMES_HOME (not from the project 62 # dir for project plugins), so that's where we opt in. 63 import os 64 hermes_home_str = os.environ.get("HERMES_HOME") 65 if hermes_home_str: 66 hermes_home = Path(hermes_home_str) 67 else: 68 hermes_home = base.parent 69 hermes_home.mkdir(parents=True, exist_ok=True) 70 cfg_path = hermes_home / "config.yaml" 71 cfg: dict = {} 72 if cfg_path.exists(): 73 try: 74 cfg = yaml.safe_load(cfg_path.read_text()) or {} 75 except Exception: 76 cfg = {} 77 plugins_cfg = cfg.setdefault("plugins", {}) 78 enabled = plugins_cfg.setdefault("enabled", []) 79 if isinstance(enabled, list) and name not in enabled: 80 enabled.append(name) 81 cfg_path.write_text(yaml.safe_dump(cfg)) 82 83 return plugin_dir 84 85 86 # ── TestPluginDiscovery ──────────────────────────────────────────────────── 87 88 89 class TestPluginDiscovery: 90 """Tests for plugin discovery from directories and entry points.""" 91 92 def test_discover_user_plugins(self, tmp_path, monkeypatch): 93 """Plugins in ~/.hermes/plugins/ are discovered.""" 94 plugins_dir = tmp_path / "hermes_test" / "plugins" 95 _make_plugin_dir(plugins_dir, "hello_plugin") 96 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 97 98 mgr = PluginManager() 99 mgr.discover_and_load() 100 101 assert "hello_plugin" in mgr._plugins 102 assert mgr._plugins["hello_plugin"].enabled 103 104 def test_discover_project_plugins(self, tmp_path, monkeypatch): 105 """Plugins in ./.hermes/plugins/ are discovered.""" 106 project_dir = tmp_path / "project" 107 project_dir.mkdir() 108 monkeypatch.chdir(project_dir) 109 monkeypatch.setenv("HERMES_ENABLE_PROJECT_PLUGINS", "true") 110 plugins_dir = project_dir / ".hermes" / "plugins" 111 _make_plugin_dir(plugins_dir, "proj_plugin") 112 113 mgr = PluginManager() 114 mgr.discover_and_load() 115 116 assert "proj_plugin" in mgr._plugins 117 assert mgr._plugins["proj_plugin"].enabled 118 119 def test_discover_project_plugins_skipped_by_default(self, tmp_path, monkeypatch): 120 """Project plugins are not discovered unless explicitly enabled.""" 121 project_dir = tmp_path / "project" 122 project_dir.mkdir() 123 monkeypatch.chdir(project_dir) 124 plugins_dir = project_dir / ".hermes" / "plugins" 125 _make_plugin_dir(plugins_dir, "proj_plugin") 126 127 mgr = PluginManager() 128 mgr.discover_and_load() 129 130 assert "proj_plugin" not in mgr._plugins 131 132 def test_discover_is_idempotent(self, tmp_path, monkeypatch): 133 """Calling discover_and_load() twice does not duplicate plugins.""" 134 plugins_dir = tmp_path / "hermes_test" / "plugins" 135 _make_plugin_dir(plugins_dir, "once_plugin") 136 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 137 138 mgr = PluginManager() 139 mgr.discover_and_load() 140 mgr.discover_and_load() # second call should no-op 141 142 # Filter out bundled plugins — they're always discovered. 143 non_bundled = { 144 n: p for n, p in mgr._plugins.items() 145 if p.manifest.source != "bundled" 146 } 147 assert len(non_bundled) == 1 148 149 def test_discover_skips_dir_without_manifest(self, tmp_path, monkeypatch): 150 """Directories without plugin.yaml are silently skipped.""" 151 plugins_dir = tmp_path / "hermes_test" / "plugins" 152 (plugins_dir / "no_manifest").mkdir(parents=True) 153 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 154 155 mgr = PluginManager() 156 mgr.discover_and_load() 157 158 # Filter out bundled plugins — they're always discovered. 159 non_bundled = { 160 n: p for n, p in mgr._plugins.items() 161 if p.manifest.source != "bundled" 162 } 163 assert len(non_bundled) == 0 164 165 def test_entry_points_scanned(self, tmp_path, monkeypatch): 166 """Entry-point based plugins are discovered (mocked).""" 167 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 168 169 fake_module = types.ModuleType("fake_ep_plugin") 170 fake_module.register = lambda ctx: None # type: ignore[attr-defined] 171 172 fake_ep = MagicMock() 173 fake_ep.name = "ep_plugin" 174 fake_ep.value = "fake_ep_plugin:register" 175 fake_ep.group = ENTRY_POINTS_GROUP 176 fake_ep.load.return_value = fake_module 177 178 def fake_entry_points(): 179 result = MagicMock() 180 result.select = MagicMock(return_value=[fake_ep]) 181 return result 182 183 with patch("importlib.metadata.entry_points", fake_entry_points): 184 mgr = PluginManager() 185 mgr.discover_and_load() 186 187 assert "ep_plugin" in mgr._plugins 188 189 190 # ── TestPluginLoading ────────────────────────────────────────────────────── 191 192 193 class TestPluginLoading: 194 """Tests for plugin module loading.""" 195 196 def test_load_missing_init(self, tmp_path, monkeypatch): 197 """Plugin dir without __init__.py records an error.""" 198 plugins_dir = tmp_path / "hermes_test" / "plugins" 199 plugin_dir = plugins_dir / "bad_plugin" 200 plugin_dir.mkdir(parents=True) 201 (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "bad_plugin"})) 202 # Explicitly enable so the loader tries to import it and hits the 203 # missing-init error. 204 hermes_home = tmp_path / "hermes_test" 205 (hermes_home / "config.yaml").write_text( 206 yaml.safe_dump({"plugins": {"enabled": ["bad_plugin"]}}) 207 ) 208 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 209 210 mgr = PluginManager() 211 mgr.discover_and_load() 212 213 assert "bad_plugin" in mgr._plugins 214 assert not mgr._plugins["bad_plugin"].enabled 215 assert mgr._plugins["bad_plugin"].error is not None 216 # Should be the missing-init error, not "not enabled". 217 assert "not enabled" not in mgr._plugins["bad_plugin"].error 218 219 def test_load_missing_register_fn(self, tmp_path, monkeypatch): 220 """Plugin without register() function records an error.""" 221 plugins_dir = tmp_path / "hermes_test" / "plugins" 222 plugin_dir = plugins_dir / "no_reg" 223 plugin_dir.mkdir(parents=True) 224 (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "no_reg"})) 225 (plugin_dir / "__init__.py").write_text("# no register function\n") 226 # Explicitly enable it so the loader actually tries to import. 227 hermes_home = tmp_path / "hermes_test" 228 (hermes_home / "config.yaml").write_text( 229 yaml.safe_dump({"plugins": {"enabled": ["no_reg"]}}) 230 ) 231 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 232 233 mgr = PluginManager() 234 mgr.discover_and_load() 235 236 assert "no_reg" in mgr._plugins 237 assert not mgr._plugins["no_reg"].enabled 238 assert "no register()" in mgr._plugins["no_reg"].error 239 240 def test_load_registers_namespace_module(self, tmp_path, monkeypatch): 241 """Directory plugins are importable under hermes_plugins.<name>.""" 242 plugins_dir = tmp_path / "hermes_test" / "plugins" 243 _make_plugin_dir(plugins_dir, "ns_plugin") 244 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 245 246 # Clean up any prior namespace module 247 sys.modules.pop("hermes_plugins.ns_plugin", None) 248 249 mgr = PluginManager() 250 mgr.discover_and_load() 251 252 assert "hermes_plugins.ns_plugin" in sys.modules 253 254 def test_user_memory_plugin_auto_coerced_to_exclusive(self, tmp_path, monkeypatch): 255 """User-installed memory plugins must NOT be loaded by the general 256 PluginManager — they belong to plugins/memory discovery. 257 258 Regression test for the mempalace crash: 259 'PluginContext' object has no attribute 'register_memory_provider' 260 261 A plugin that calls ``ctx.register_memory_provider`` in its 262 ``__init__.py`` should be auto-detected and treated as 263 ``kind: exclusive`` so the general loader records the manifest but 264 does not import/register() it. The real activation happens through 265 ``plugins/memory/__init__.py`` via ``memory.provider`` config. 266 """ 267 plugins_dir = tmp_path / "hermes_test" / "plugins" 268 plugin_dir = plugins_dir / "mempalace" 269 plugin_dir.mkdir(parents=True) 270 # No explicit `kind:` — the heuristic should kick in. 271 (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "mempalace"})) 272 (plugin_dir / "__init__.py").write_text( 273 "class MemPalaceProvider:\n" 274 " pass\n" 275 "def register(ctx):\n" 276 " ctx.register_memory_provider('mempalace', MemPalaceProvider)\n" 277 ) 278 # Even if the user explicitly enables it in config, the loader 279 # should still treat it as exclusive and skip general loading. 280 hermes_home = tmp_path / "hermes_test" 281 (hermes_home / "config.yaml").write_text( 282 yaml.safe_dump({"plugins": {"enabled": ["mempalace"]}}) 283 ) 284 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 285 286 mgr = PluginManager() 287 mgr.discover_and_load() 288 289 assert "mempalace" in mgr._plugins 290 entry = mgr._plugins["mempalace"] 291 assert entry.manifest.kind == "exclusive", ( 292 f"Expected auto-coerced kind='exclusive', got {entry.manifest.kind}" 293 ) 294 # Not loaded by general manager (no register() call, no AttributeError). 295 assert not entry.enabled 296 assert entry.module is None 297 assert "exclusive" in (entry.error or "").lower() 298 299 def test_explicit_standalone_kind_not_coerced(self, tmp_path, monkeypatch): 300 """If a plugin explicitly declares ``kind: standalone`` in its 301 manifest, the memory-provider heuristic must NOT override it — 302 even if the source happens to mention ``MemoryProvider``. 303 """ 304 plugins_dir = tmp_path / "hermes_test" / "plugins" 305 plugin_dir = plugins_dir / "not_memory" 306 plugin_dir.mkdir(parents=True) 307 (plugin_dir / "plugin.yaml").write_text( 308 yaml.dump({"name": "not_memory", "kind": "standalone"}) 309 ) 310 (plugin_dir / "__init__.py").write_text( 311 "# This plugin inspects MemoryProvider docs but isn't one.\n" 312 "def register(ctx):\n pass\n" 313 ) 314 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 315 316 mgr = PluginManager() 317 mgr.discover_and_load() 318 319 assert mgr._plugins["not_memory"].manifest.kind == "standalone" 320 321 322 # ── TestPluginHooks ──────────────────────────────────────────────────────── 323 324 325 class TestPluginHooks: 326 """Tests for lifecycle hook registration and invocation.""" 327 328 def test_valid_hooks_include_request_scoped_api_hooks(self): 329 assert "pre_api_request" in VALID_HOOKS 330 assert "post_api_request" in VALID_HOOKS 331 assert "transform_terminal_output" in VALID_HOOKS 332 assert "transform_tool_result" in VALID_HOOKS 333 334 def test_valid_hooks_include_pre_gateway_dispatch(self): 335 assert "pre_gateway_dispatch" in VALID_HOOKS 336 337 def test_pre_gateway_dispatch_collects_action_dicts(self, tmp_path, monkeypatch): 338 """pre_gateway_dispatch callbacks return action dicts (skip/rewrite/allow).""" 339 plugins_dir = tmp_path / "hermes_test" / "plugins" 340 _make_plugin_dir( 341 plugins_dir, "predispatch_plugin", 342 register_body=( 343 'ctx.register_hook("pre_gateway_dispatch", ' 344 'lambda **kw: {"action": "skip", "reason": "test"})' 345 ), 346 ) 347 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 348 349 mgr = PluginManager() 350 mgr.discover_and_load() 351 352 results = mgr.invoke_hook( 353 "pre_gateway_dispatch", 354 event=object(), 355 gateway=object(), 356 session_store=object(), 357 ) 358 assert len(results) == 1 359 assert results[0] == {"action": "skip", "reason": "test"} 360 361 def test_register_and_invoke_hook(self, tmp_path, monkeypatch): 362 """Registered hooks are called on invoke_hook().""" 363 plugins_dir = tmp_path / "hermes_test" / "plugins" 364 _make_plugin_dir( 365 plugins_dir, "hook_plugin", 366 register_body='ctx.register_hook("pre_tool_call", lambda **kw: None)', 367 ) 368 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 369 370 mgr = PluginManager() 371 mgr.discover_and_load() 372 373 # Should not raise 374 mgr.invoke_hook("pre_tool_call", tool_name="test", args={}, task_id="t1") 375 376 def test_hook_exception_does_not_propagate(self, tmp_path, monkeypatch): 377 """A hook callback that raises does NOT crash the caller.""" 378 plugins_dir = tmp_path / "hermes_test" / "plugins" 379 _make_plugin_dir( 380 plugins_dir, "bad_hook", 381 register_body='ctx.register_hook("post_tool_call", lambda **kw: 1/0)', 382 ) 383 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 384 385 mgr = PluginManager() 386 mgr.discover_and_load() 387 388 # Should not raise despite 1/0 389 mgr.invoke_hook("post_tool_call", tool_name="x", args={}, result="r", task_id="") 390 391 def test_hook_return_values_collected(self, tmp_path, monkeypatch): 392 """invoke_hook() collects non-None return values from callbacks.""" 393 plugins_dir = tmp_path / "hermes_test" / "plugins" 394 _make_plugin_dir( 395 plugins_dir, "ctx_plugin", 396 register_body=( 397 'ctx.register_hook("pre_llm_call", ' 398 'lambda **kw: {"context": "memory from plugin"})' 399 ), 400 ) 401 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 402 403 mgr = PluginManager() 404 mgr.discover_and_load() 405 406 results = mgr.invoke_hook("pre_llm_call", session_id="s1", user_message="hi", 407 conversation_history=[], is_first_turn=True, model="test") 408 assert len(results) == 1 409 assert results[0] == {"context": "memory from plugin"} 410 411 def test_hook_none_returns_excluded(self, tmp_path, monkeypatch): 412 """invoke_hook() excludes None returns from the result list.""" 413 plugins_dir = tmp_path / "hermes_test" / "plugins" 414 _make_plugin_dir( 415 plugins_dir, "none_hook", 416 register_body='ctx.register_hook("post_llm_call", lambda **kw: None)', 417 ) 418 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 419 420 mgr = PluginManager() 421 mgr.discover_and_load() 422 423 results = mgr.invoke_hook("post_llm_call", session_id="s1", 424 user_message="hi", assistant_response="bye", model="test") 425 assert results == [] 426 427 def test_request_hooks_are_invokeable(self, tmp_path, monkeypatch): 428 plugins_dir = tmp_path / "hermes_test" / "plugins" 429 _make_plugin_dir( 430 plugins_dir, "request_hook", 431 register_body=( 432 'ctx.register_hook("pre_api_request", ' 433 'lambda **kw: {"seen": kw.get("api_call_count"), ' 434 '"mc": kw.get("message_count"), "tc": kw.get("tool_count")})' 435 ), 436 ) 437 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 438 439 mgr = PluginManager() 440 mgr.discover_and_load() 441 442 results = mgr.invoke_hook( 443 "pre_api_request", 444 session_id="s1", 445 task_id="t1", 446 model="test", 447 api_call_count=2, 448 message_count=5, 449 tool_count=3, 450 approx_input_tokens=100, 451 request_char_count=400, 452 max_tokens=8192, 453 ) 454 assert results == [{"seen": 2, "mc": 5, "tc": 3}] 455 456 def test_transform_terminal_output_hook_can_be_registered_and_invoked(self, tmp_path, monkeypatch): 457 plugins_dir = tmp_path / "hermes_test" / "plugins" 458 _make_plugin_dir( 459 plugins_dir, "transform_hook", 460 register_body=( 461 'ctx.register_hook("transform_terminal_output", ' 462 'lambda **kw: f"{kw[\'command\']}|{kw[\'returncode\']}|{kw[\'env_type\']}|{kw[\'task_id\']}|{len(kw[\'output\'])}")' 463 ), 464 ) 465 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 466 467 mgr = PluginManager() 468 mgr.discover_and_load() 469 470 results = mgr.invoke_hook( 471 "transform_terminal_output", 472 command="echo hello", 473 output="abcdef", 474 returncode=7, 475 task_id="task-1", 476 env_type="local", 477 ) 478 assert results == ["echo hello|7|local|task-1|6"] 479 480 def test_invalid_hook_name_warns(self, tmp_path, monkeypatch, caplog): 481 """Registering an unknown hook name logs a warning.""" 482 plugins_dir = tmp_path / "hermes_test" / "plugins" 483 _make_plugin_dir( 484 plugins_dir, "warn_plugin", 485 register_body='ctx.register_hook("on_banana", lambda **kw: None)', 486 ) 487 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 488 489 with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): 490 mgr = PluginManager() 491 mgr.discover_and_load() 492 493 assert any("on_banana" in record.message for record in caplog.records) 494 495 496 class TestPreToolCallBlocking: 497 """Tests for the pre_tool_call block directive helper.""" 498 499 def test_block_message_returned_for_valid_directive(self, monkeypatch): 500 monkeypatch.setattr( 501 "hermes_cli.plugins.invoke_hook", 502 lambda hook_name, **kwargs: [{"action": "block", "message": "blocked by plugin"}], 503 ) 504 assert get_pre_tool_call_block_message("todo", {}, task_id="t1") == "blocked by plugin" 505 506 def test_invalid_returns_are_ignored(self, monkeypatch): 507 """Various malformed hook returns should not trigger a block.""" 508 monkeypatch.setattr( 509 "hermes_cli.plugins.invoke_hook", 510 lambda hook_name, **kwargs: [ 511 "block", # not a dict 512 123, # not a dict 513 {"action": "block"}, # missing message 514 {"action": "deny", "message": "nope"}, # wrong action 515 {"message": "missing action"}, # no action key 516 {"action": "block", "message": 123}, # message not str 517 ], 518 ) 519 assert get_pre_tool_call_block_message("todo", {}, task_id="t1") is None 520 521 def test_none_when_no_hooks(self, monkeypatch): 522 monkeypatch.setattr( 523 "hermes_cli.plugins.invoke_hook", 524 lambda hook_name, **kwargs: [], 525 ) 526 assert get_pre_tool_call_block_message("web_search", {"q": "test"}) is None 527 528 def test_first_valid_block_wins(self, monkeypatch): 529 monkeypatch.setattr( 530 "hermes_cli.plugins.invoke_hook", 531 lambda hook_name, **kwargs: [ 532 {"action": "allow"}, 533 {"action": "block", "message": "first blocker"}, 534 {"action": "block", "message": "second blocker"}, 535 ], 536 ) 537 assert get_pre_tool_call_block_message("terminal", {}) == "first blocker" 538 539 540 # ── TestPluginContext ────────────────────────────────────────────────────── 541 542 543 class TestPluginContext: 544 """Tests for the PluginContext facade.""" 545 546 def test_register_tool_adds_to_registry(self, tmp_path, monkeypatch): 547 """PluginContext.register_tool() puts the tool in the global registry.""" 548 plugins_dir = tmp_path / "hermes_test" / "plugins" 549 plugin_dir = plugins_dir / "tool_plugin" 550 plugin_dir.mkdir(parents=True) 551 (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "tool_plugin"})) 552 (plugin_dir / "__init__.py").write_text( 553 'def register(ctx):\n' 554 ' ctx.register_tool(\n' 555 ' name="plugin_echo",\n' 556 ' toolset="plugin_tool_plugin",\n' 557 ' schema={"name": "plugin_echo", "description": "Echo", "parameters": {"type": "object", "properties": {}}},\n' 558 ' handler=lambda args, **kw: "echo",\n' 559 ' )\n' 560 ) 561 hermes_home = tmp_path / "hermes_test" 562 (hermes_home / "config.yaml").write_text( 563 yaml.safe_dump({"plugins": {"enabled": ["tool_plugin"]}}) 564 ) 565 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 566 567 mgr = PluginManager() 568 mgr.discover_and_load() 569 570 assert "plugin_echo" in mgr._plugin_tool_names 571 572 from tools.registry import registry 573 assert "plugin_echo" in registry._tools 574 575 576 # ── TestPluginToolVisibility ─────────────────────────────────────────────── 577 578 579 class TestPluginToolVisibility: 580 """Plugin-registered tools appear in get_tool_definitions().""" 581 582 def test_plugin_tools_in_definitions(self, tmp_path, monkeypatch): 583 """Plugin tools are included when their toolset is in enabled_toolsets.""" 584 import hermes_cli.plugins as plugins_mod 585 586 plugins_dir = tmp_path / "hermes_test" / "plugins" 587 plugin_dir = plugins_dir / "vis_plugin" 588 plugin_dir.mkdir(parents=True) 589 (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "vis_plugin"})) 590 (plugin_dir / "__init__.py").write_text( 591 'def register(ctx):\n' 592 ' ctx.register_tool(\n' 593 ' name="vis_tool",\n' 594 ' toolset="plugin_vis_plugin",\n' 595 ' schema={"name": "vis_tool", "description": "Visible", "parameters": {"type": "object", "properties": {}}},\n' 596 ' handler=lambda args, **kw: "ok",\n' 597 ' )\n' 598 ) 599 hermes_home = tmp_path / "hermes_test" 600 (hermes_home / "config.yaml").write_text( 601 yaml.safe_dump({"plugins": {"enabled": ["vis_plugin"]}}) 602 ) 603 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 604 605 mgr = PluginManager() 606 mgr.discover_and_load() 607 monkeypatch.setattr(plugins_mod, "_plugin_manager", mgr) 608 609 from model_tools import get_tool_definitions 610 611 # Plugin tools are included when their toolset is explicitly enabled 612 tools = get_tool_definitions(enabled_toolsets=["terminal", "plugin_vis_plugin"], quiet_mode=True) 613 tool_names = [t["function"]["name"] for t in tools] 614 assert "vis_tool" in tool_names 615 616 # Plugin tools are excluded when only other toolsets are enabled 617 tools2 = get_tool_definitions(enabled_toolsets=["terminal"], quiet_mode=True) 618 tool_names2 = [t["function"]["name"] for t in tools2] 619 assert "vis_tool" not in tool_names2 620 621 # Plugin tools are included when no toolset filter is active (all enabled) 622 tools3 = get_tool_definitions(quiet_mode=True) 623 tool_names3 = [t["function"]["name"] for t in tools3] 624 assert "vis_tool" in tool_names3 625 626 627 # ── TestPluginManagerList ────────────────────────────────────────────────── 628 629 630 class TestPluginManagerList: 631 """Tests for PluginManager.list_plugins().""" 632 633 def test_list_empty(self): 634 """Empty manager returns empty list.""" 635 mgr = PluginManager() 636 assert mgr.list_plugins() == [] 637 638 def test_list_returns_sorted(self, tmp_path, monkeypatch): 639 """list_plugins() returns results sorted by key.""" 640 plugins_dir = tmp_path / "hermes_test" / "plugins" 641 _make_plugin_dir(plugins_dir, "zulu") 642 _make_plugin_dir(plugins_dir, "alpha") 643 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 644 645 mgr = PluginManager() 646 mgr.discover_and_load() 647 648 listing = mgr.list_plugins() 649 # list_plugins sorts by key (path-derived, e.g. ``image_gen/openai``), 650 # not by display name, so that category plugins group together. 651 keys = [p["key"] for p in listing] 652 assert keys == sorted(keys) 653 654 def test_list_with_plugins(self, tmp_path, monkeypatch): 655 """list_plugins() returns info dicts for each discovered plugin.""" 656 plugins_dir = tmp_path / "hermes_test" / "plugins" 657 _make_plugin_dir(plugins_dir, "alpha") 658 _make_plugin_dir(plugins_dir, "beta") 659 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 660 661 mgr = PluginManager() 662 mgr.discover_and_load() 663 664 listing = mgr.list_plugins() 665 names = [p["name"] for p in listing] 666 assert "alpha" in names 667 assert "beta" in names 668 for p in listing: 669 assert "enabled" in p 670 assert "tools" in p 671 assert "hooks" in p 672 673 674 675 class TestPreLlmCallTargetRouting: 676 """Tests for pre_llm_call hook return format with target-aware routing. 677 678 The routing logic lives in run_agent.py, but the return format is collected 679 by invoke_hook(). These tests verify the return format works correctly and 680 that downstream code can route based on the 'target' key. 681 """ 682 683 def _make_pre_llm_plugin(self, plugins_dir, name, return_expr): 684 """Create a plugin that returns a specific value from pre_llm_call.""" 685 _make_plugin_dir( 686 plugins_dir, name, 687 register_body=( 688 f'ctx.register_hook("pre_llm_call", lambda **kw: {return_expr})' 689 ), 690 ) 691 692 def test_context_dict_returned(self, tmp_path, monkeypatch): 693 """Plugin returning a context dict is collected by invoke_hook.""" 694 plugins_dir = tmp_path / "hermes_test" / "plugins" 695 self._make_pre_llm_plugin( 696 plugins_dir, "basic_plugin", 697 '{"context": "basic context"}', 698 ) 699 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 700 701 mgr = PluginManager() 702 mgr.discover_and_load() 703 704 results = mgr.invoke_hook( 705 "pre_llm_call", session_id="s1", user_message="hi", 706 conversation_history=[], is_first_turn=True, model="test", 707 ) 708 assert len(results) == 1 709 assert results[0]["context"] == "basic context" 710 assert "target" not in results[0] 711 712 def test_plain_string_return(self, tmp_path, monkeypatch): 713 """Plain string returns are collected as-is (routing treats them as user_message).""" 714 plugins_dir = tmp_path / "hermes_test" / "plugins" 715 self._make_pre_llm_plugin( 716 plugins_dir, "str_plugin", 717 '"plain string context"', 718 ) 719 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 720 721 mgr = PluginManager() 722 mgr.discover_and_load() 723 724 results = mgr.invoke_hook( 725 "pre_llm_call", session_id="s1", user_message="hi", 726 conversation_history=[], is_first_turn=True, model="test", 727 ) 728 assert len(results) == 1 729 assert results[0] == "plain string context" 730 731 def test_multiple_plugins_context_collected(self, tmp_path, monkeypatch): 732 """Multiple plugins returning context are all collected.""" 733 plugins_dir = tmp_path / "hermes_test" / "plugins" 734 self._make_pre_llm_plugin( 735 plugins_dir, "aaa_memory", 736 '{"context": "memory context"}', 737 ) 738 self._make_pre_llm_plugin( 739 plugins_dir, "bbb_guardrail", 740 '{"context": "guardrail text"}', 741 ) 742 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 743 744 mgr = PluginManager() 745 mgr.discover_and_load() 746 747 results = mgr.invoke_hook( 748 "pre_llm_call", session_id="s1", user_message="hi", 749 conversation_history=[], is_first_turn=True, model="test", 750 ) 751 assert len(results) == 2 752 contexts = [r["context"] for r in results] 753 assert "memory context" in contexts 754 assert "guardrail text" in contexts 755 756 def test_routing_logic_all_to_user_message(self, tmp_path, monkeypatch): 757 """Simulate the routing logic from run_agent.py. 758 759 All plugin context — dicts and plain strings — ends up in a single 760 user message context string. There is no system_prompt target. 761 """ 762 plugins_dir = tmp_path / "hermes_test" / "plugins" 763 self._make_pre_llm_plugin( 764 plugins_dir, "aaa_mem", 765 '{"context": "memory A"}', 766 ) 767 self._make_pre_llm_plugin( 768 plugins_dir, "bbb_guard", 769 '{"context": "rule B"}', 770 ) 771 self._make_pre_llm_plugin( 772 plugins_dir, "ccc_plain", 773 '"plain text C"', 774 ) 775 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 776 777 mgr = PluginManager() 778 mgr.discover_and_load() 779 780 results = mgr.invoke_hook( 781 "pre_llm_call", session_id="s1", user_message="hi", 782 conversation_history=[], is_first_turn=True, model="test", 783 ) 784 785 # Replicate run_agent.py routing logic — everything goes to user msg 786 _ctx_parts = [] 787 for r in results: 788 if isinstance(r, dict) and r.get("context"): 789 _ctx_parts.append(str(r["context"])) 790 elif isinstance(r, str) and r.strip(): 791 _ctx_parts.append(r) 792 793 assert _ctx_parts == ["memory A", "rule B", "plain text C"] 794 _plugin_user_context = "\n\n".join(_ctx_parts) 795 assert "memory A" in _plugin_user_context 796 assert "rule B" in _plugin_user_context 797 assert "plain text C" in _plugin_user_context 798 799 800 # ── TestPluginCommands ──────────────────────────────────────────────────── 801 802 803 class TestPluginCommands: 804 """Tests for plugin slash command registration via register_command().""" 805 806 def test_register_command_basic(self): 807 """register_command() stores handler, description, and plugin name.""" 808 mgr = PluginManager() 809 manifest = PluginManifest(name="test-plugin", source="user") 810 ctx = PluginContext(manifest, mgr) 811 812 handler = lambda args: f"echo {args}" 813 ctx.register_command("mycmd", handler, description="My custom command") 814 815 assert "mycmd" in mgr._plugin_commands 816 entry = mgr._plugin_commands["mycmd"] 817 assert entry["handler"] is handler 818 assert entry["description"] == "My custom command" 819 assert entry["plugin"] == "test-plugin" 820 # args_hint defaults to empty string when not passed. 821 assert entry["args_hint"] == "" 822 823 def test_register_command_with_args_hint(self): 824 """args_hint is stored and surfaced for gateway-native UI registration.""" 825 mgr = PluginManager() 826 manifest = PluginManifest(name="test-plugin", source="user") 827 ctx = PluginContext(manifest, mgr) 828 829 ctx.register_command( 830 "metricas", 831 lambda a: a, 832 description="Metrics dashboard", 833 args_hint="dias:7 formato:json", 834 ) 835 836 entry = mgr._plugin_commands["metricas"] 837 assert entry["args_hint"] == "dias:7 formato:json" 838 839 def test_register_command_args_hint_whitespace_trimmed(self): 840 """args_hint leading/trailing whitespace is stripped.""" 841 mgr = PluginManager() 842 manifest = PluginManifest(name="test-plugin", source="user") 843 ctx = PluginContext(manifest, mgr) 844 845 ctx.register_command("foo", lambda a: a, args_hint=" <file> ") 846 assert mgr._plugin_commands["foo"]["args_hint"] == "<file>" 847 848 def test_register_command_normalizes_name(self): 849 """Names are lowercased, stripped, and leading slashes removed.""" 850 mgr = PluginManager() 851 manifest = PluginManifest(name="test-plugin", source="user") 852 ctx = PluginContext(manifest, mgr) 853 854 ctx.register_command("/MyCmd ", lambda a: a, description="test") 855 assert "mycmd" in mgr._plugin_commands 856 assert "/MyCmd " not in mgr._plugin_commands 857 858 def test_register_command_empty_name_rejected(self, caplog): 859 """Empty name after normalization is rejected with a warning.""" 860 mgr = PluginManager() 861 manifest = PluginManifest(name="test-plugin", source="user") 862 ctx = PluginContext(manifest, mgr) 863 864 with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): 865 ctx.register_command("", lambda a: a) 866 assert len(mgr._plugin_commands) == 0 867 assert "empty name" in caplog.text 868 869 def test_register_command_builtin_conflict_rejected(self, caplog): 870 """Commands that conflict with built-in names are rejected.""" 871 mgr = PluginManager() 872 manifest = PluginManifest(name="test-plugin", source="user") 873 ctx = PluginContext(manifest, mgr) 874 875 with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): 876 ctx.register_command("help", lambda a: a) 877 assert "help" not in mgr._plugin_commands 878 assert "conflicts" in caplog.text.lower() 879 880 def test_register_command_default_description(self): 881 """Missing description defaults to 'Plugin command'.""" 882 mgr = PluginManager() 883 manifest = PluginManifest(name="test-plugin", source="user") 884 ctx = PluginContext(manifest, mgr) 885 886 ctx.register_command("status-cmd", lambda a: a) 887 assert mgr._plugin_commands["status-cmd"]["description"] == "Plugin command" 888 889 def test_get_plugin_command_handler_found(self): 890 """get_plugin_command_handler() returns the handler for a registered command.""" 891 mgr = PluginManager() 892 manifest = PluginManifest(name="test-plugin", source="user") 893 ctx = PluginContext(manifest, mgr) 894 895 handler = lambda args: f"result: {args}" 896 ctx.register_command("mycmd", handler, description="test") 897 898 with patch("hermes_cli.plugins._plugin_manager", mgr): 899 result = get_plugin_command_handler("mycmd") 900 assert result is handler 901 902 def test_get_plugin_command_handler_not_found(self): 903 """get_plugin_command_handler() returns None for unregistered commands.""" 904 mgr = PluginManager() 905 with patch("hermes_cli.plugins._plugin_manager", mgr): 906 assert get_plugin_command_handler("nonexistent") is None 907 908 def test_get_plugin_commands_returns_dict(self): 909 """get_plugin_commands() returns the full commands dict.""" 910 mgr = PluginManager() 911 manifest = PluginManifest(name="test-plugin", source="user") 912 ctx = PluginContext(manifest, mgr) 913 ctx.register_command("cmd-a", lambda a: a, description="A") 914 ctx.register_command("cmd-b", lambda a: a, description="B") 915 916 with patch("hermes_cli.plugins._plugin_manager", mgr): 917 cmds = get_plugin_commands() 918 assert "cmd-a" in cmds 919 assert "cmd-b" in cmds 920 assert cmds["cmd-a"]["description"] == "A" 921 922 def test_get_plugin_command_handler_discovers_plugins_lazily(self, tmp_path, monkeypatch): 923 """Handler lookup should work before any explicit discover_plugins() call.""" 924 plugins_dir = tmp_path / "hermes_test" / "plugins" 925 _make_plugin_dir( 926 plugins_dir, 927 "cmd-plugin", 928 register_body='ctx.register_command("lazycmd", lambda a: f"ok:{a}", description="Lazy")', 929 ) 930 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 931 932 import hermes_cli.plugins as plugins_mod 933 934 with patch.object(plugins_mod, "_plugin_manager", None): 935 handler = get_plugin_command_handler("lazycmd") 936 assert handler is not None 937 assert handler("x") == "ok:x" 938 939 def test_get_plugin_commands_discovers_plugins_lazily(self, tmp_path, monkeypatch): 940 """Command listing should trigger plugin discovery on first access.""" 941 plugins_dir = tmp_path / "hermes_test" / "plugins" 942 _make_plugin_dir( 943 plugins_dir, 944 "cmd-plugin", 945 register_body='ctx.register_command("lazycmd", lambda a: a, description="Lazy")', 946 ) 947 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 948 949 import hermes_cli.plugins as plugins_mod 950 951 with patch.object(plugins_mod, "_plugin_manager", None): 952 cmds = get_plugin_commands() 953 assert "lazycmd" in cmds 954 assert cmds["lazycmd"]["description"] == "Lazy" 955 956 def test_get_plugin_context_engine_discovers_plugins_lazily(self, tmp_path, monkeypatch): 957 """Context engine lookup should work before any explicit discover_plugins() call.""" 958 hermes_home = tmp_path / "hermes_test" 959 plugins_dir = hermes_home / "plugins" 960 plugin_dir = plugins_dir / "engine-plugin" 961 plugin_dir.mkdir(parents=True, exist_ok=True) 962 (plugin_dir / "plugin.yaml").write_text( 963 yaml.dump({ 964 "name": "engine-plugin", 965 "version": "0.1.0", 966 "description": "Test engine plugin", 967 }) 968 ) 969 (plugin_dir / "__init__.py").write_text( 970 "from agent.context_engine import ContextEngine\n\n" 971 "class StubEngine(ContextEngine):\n" 972 " @property\n" 973 " def name(self):\n" 974 " return 'stub-engine'\n\n" 975 " def update_from_response(self, usage):\n" 976 " return None\n\n" 977 " def should_compress(self, prompt_tokens):\n" 978 " return False\n\n" 979 " def compress(self, messages, current_tokens):\n" 980 " return messages\n\n" 981 "def register(ctx):\n" 982 " ctx.register_context_engine(StubEngine())\n" 983 ) 984 # Opt-in: plugins are opt-in by default, so enable in config.yaml 985 (hermes_home / "config.yaml").write_text( 986 yaml.safe_dump({"plugins": {"enabled": ["engine-plugin"]}}) 987 ) 988 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 989 990 import hermes_cli.plugins as plugins_mod 991 992 with patch.object(plugins_mod, "_plugin_manager", None): 993 engine = plugins_mod.get_plugin_context_engine() 994 assert engine is not None 995 assert engine.name == "stub-engine" 996 997 def test_commands_tracked_on_loaded_plugin(self, tmp_path, monkeypatch): 998 """Commands registered during discover_and_load() are tracked on LoadedPlugin.""" 999 plugins_dir = tmp_path / "hermes_test" / "plugins" 1000 _make_plugin_dir( 1001 plugins_dir, "cmd-plugin", 1002 register_body=( 1003 'ctx.register_command("mycmd", lambda a: "ok", description="Test")' 1004 ), 1005 ) 1006 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 1007 1008 mgr = PluginManager() 1009 mgr.discover_and_load() 1010 1011 loaded = mgr._plugins["cmd-plugin"] 1012 assert loaded.enabled 1013 assert "mycmd" in loaded.commands_registered 1014 1015 def test_commands_in_list_plugins_output(self, tmp_path, monkeypatch): 1016 """list_plugins() includes command count.""" 1017 plugins_dir = tmp_path / "hermes_test" / "plugins" 1018 # Set HERMES_HOME BEFORE _make_plugin_dir so auto-enable targets 1019 # the right config.yaml. 1020 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) 1021 _make_plugin_dir( 1022 plugins_dir, "cmd-plugin", 1023 register_body=( 1024 'ctx.register_command("mycmd", lambda a: "ok", description="Test")' 1025 ), 1026 ) 1027 1028 mgr = PluginManager() 1029 mgr.discover_and_load() 1030 1031 info = mgr.list_plugins() 1032 # Filter out bundled plugins — they're always discovered. 1033 cmd_info = [p for p in info if p["name"] == "cmd-plugin"] 1034 assert len(cmd_info) == 1 1035 assert cmd_info[0]["commands"] == 1 1036 1037 def test_handler_receives_raw_args(self): 1038 """The handler is called with the raw argument string.""" 1039 mgr = PluginManager() 1040 manifest = PluginManifest(name="test-plugin", source="user") 1041 ctx = PluginContext(manifest, mgr) 1042 1043 received = [] 1044 ctx.register_command("echo", lambda args: received.append(args) or "ok") 1045 1046 handler = mgr._plugin_commands["echo"]["handler"] 1047 handler("hello world") 1048 assert received == ["hello world"] 1049 1050 def test_multiple_plugins_register_different_commands(self): 1051 """Multiple plugins can each register their own commands.""" 1052 mgr = PluginManager() 1053 1054 for plugin_name, cmd_name in [("plugin-a", "cmd-a"), ("plugin-b", "cmd-b")]: 1055 manifest = PluginManifest(name=plugin_name, source="user") 1056 ctx = PluginContext(manifest, mgr) 1057 ctx.register_command(cmd_name, lambda a: a, description=f"From {plugin_name}") 1058 1059 assert "cmd-a" in mgr._plugin_commands 1060 assert "cmd-b" in mgr._plugin_commands 1061 assert mgr._plugin_commands["cmd-a"]["plugin"] == "plugin-a" 1062 assert mgr._plugin_commands["cmd-b"]["plugin"] == "plugin-b" 1063 1064 1065 class TestPluginCommandResultResolution: 1066 def test_returns_sync_values_unchanged(self): 1067 assert resolve_plugin_command_result("ok") == "ok" 1068 1069 def test_awaits_async_result_without_running_loop(self): 1070 async def _handler(): 1071 return "async-ok" 1072 1073 assert resolve_plugin_command_result(_handler()) == "async-ok" 1074 1075 def test_awaits_async_result_with_running_loop(self, monkeypatch): 1076 class _Loop: 1077 pass 1078 1079 async def _handler(): 1080 return "threaded-ok" 1081 1082 monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop()) 1083 assert resolve_plugin_command_result(_handler()) == "threaded-ok" 1084 1085 def test_running_loop_timeout_does_not_hang_forever(self, monkeypatch): 1086 """Threaded path must abort a hung async handler instead of blocking the caller.""" 1087 import asyncio as _asyncio 1088 1089 class _Loop: 1090 pass 1091 1092 async def _slow_handler(): 1093 await _asyncio.sleep(10) 1094 return "should-not-reach" 1095 1096 monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop()) 1097 monkeypatch.setattr("hermes_cli.plugins._PLUGIN_COMMAND_AWAIT_TIMEOUT_SECS", 0.1) 1098 1099 import pytest 1100 with pytest.raises(TimeoutError): 1101 resolve_plugin_command_result(_slow_handler()) 1102 1103 1104 # ── TestPluginDispatchTool ──────────────────────────────────────────────── 1105 1106 1107 class TestPluginDispatchTool: 1108 """Tests for PluginContext.dispatch_tool() — tool dispatch with agent context.""" 1109 1110 def test_dispatch_tool_calls_registry(self): 1111 """dispatch_tool() delegates to registry.dispatch().""" 1112 mgr = PluginManager() 1113 manifest = PluginManifest(name="test-plugin", source="user") 1114 ctx = PluginContext(manifest, mgr) 1115 1116 mock_registry = MagicMock() 1117 mock_registry.dispatch.return_value = '{"result": "ok"}' 1118 1119 with patch("hermes_cli.plugins.PluginContext.dispatch_tool.__module__", "hermes_cli.plugins"): 1120 with patch.dict("sys.modules", {}): 1121 with patch("tools.registry.registry", mock_registry): 1122 result = ctx.dispatch_tool("web_search", {"query": "test"}) 1123 1124 assert result == '{"result": "ok"}' 1125 1126 def test_dispatch_tool_injects_parent_agent_from_cli_ref(self): 1127 """When _cli_ref has an agent, it's passed as parent_agent.""" 1128 mgr = PluginManager() 1129 manifest = PluginManifest(name="test-plugin", source="user") 1130 ctx = PluginContext(manifest, mgr) 1131 1132 mock_agent = MagicMock() 1133 mock_cli = MagicMock() 1134 mock_cli.agent = mock_agent 1135 mgr._cli_ref = mock_cli 1136 1137 mock_registry = MagicMock() 1138 mock_registry.dispatch.return_value = '{"ok": true}' 1139 1140 with patch("tools.registry.registry", mock_registry): 1141 ctx.dispatch_tool("delegate_task", {"goal": "test"}) 1142 1143 mock_registry.dispatch.assert_called_once() 1144 call_kwargs = mock_registry.dispatch.call_args 1145 assert call_kwargs[1].get("parent_agent") is mock_agent 1146 1147 def test_dispatch_tool_no_parent_agent_when_no_cli_ref(self): 1148 """When _cli_ref is None (gateway mode), no parent_agent is injected.""" 1149 mgr = PluginManager() 1150 manifest = PluginManifest(name="test-plugin", source="user") 1151 ctx = PluginContext(manifest, mgr) 1152 mgr._cli_ref = None 1153 1154 mock_registry = MagicMock() 1155 mock_registry.dispatch.return_value = '{"ok": true}' 1156 1157 with patch("tools.registry.registry", mock_registry): 1158 ctx.dispatch_tool("delegate_task", {"goal": "test"}) 1159 1160 call_kwargs = mock_registry.dispatch.call_args 1161 assert "parent_agent" not in call_kwargs[1] 1162 1163 def test_dispatch_tool_no_parent_agent_when_agent_is_none(self): 1164 """When cli_ref exists but agent is None (not yet initialized), skip parent_agent.""" 1165 mgr = PluginManager() 1166 manifest = PluginManifest(name="test-plugin", source="user") 1167 ctx = PluginContext(manifest, mgr) 1168 1169 mock_cli = MagicMock() 1170 mock_cli.agent = None 1171 mgr._cli_ref = mock_cli 1172 1173 mock_registry = MagicMock() 1174 mock_registry.dispatch.return_value = '{"ok": true}' 1175 1176 with patch("tools.registry.registry", mock_registry): 1177 ctx.dispatch_tool("delegate_task", {"goal": "test"}) 1178 1179 call_kwargs = mock_registry.dispatch.call_args 1180 assert "parent_agent" not in call_kwargs[1] 1181 1182 def test_dispatch_tool_respects_explicit_parent_agent(self): 1183 """Explicit parent_agent kwarg is not overwritten by _cli_ref.agent.""" 1184 mgr = PluginManager() 1185 manifest = PluginManifest(name="test-plugin", source="user") 1186 ctx = PluginContext(manifest, mgr) 1187 1188 cli_agent = MagicMock(name="cli_agent") 1189 mock_cli = MagicMock() 1190 mock_cli.agent = cli_agent 1191 mgr._cli_ref = mock_cli 1192 1193 explicit_agent = MagicMock(name="explicit_agent") 1194 1195 mock_registry = MagicMock() 1196 mock_registry.dispatch.return_value = '{"ok": true}' 1197 1198 with patch("tools.registry.registry", mock_registry): 1199 ctx.dispatch_tool("delegate_task", {"goal": "test"}, parent_agent=explicit_agent) 1200 1201 call_kwargs = mock_registry.dispatch.call_args 1202 assert call_kwargs[1]["parent_agent"] is explicit_agent 1203 1204 def test_dispatch_tool_forwards_extra_kwargs(self): 1205 """Extra kwargs are forwarded to registry.dispatch().""" 1206 mgr = PluginManager() 1207 manifest = PluginManifest(name="test-plugin", source="user") 1208 ctx = PluginContext(manifest, mgr) 1209 mgr._cli_ref = None 1210 1211 mock_registry = MagicMock() 1212 mock_registry.dispatch.return_value = '{"ok": true}' 1213 1214 with patch("tools.registry.registry", mock_registry): 1215 ctx.dispatch_tool("some_tool", {"x": 1}, task_id="test-123") 1216 1217 call_kwargs = mock_registry.dispatch.call_args 1218 assert call_kwargs[1]["task_id"] == "test-123" 1219 1220 def test_dispatch_tool_returns_json_string(self): 1221 """dispatch_tool() returns the raw JSON string from the registry.""" 1222 mgr = PluginManager() 1223 manifest = PluginManifest(name="test-plugin", source="user") 1224 ctx = PluginContext(manifest, mgr) 1225 mgr._cli_ref = None 1226 1227 mock_registry = MagicMock() 1228 mock_registry.dispatch.return_value = '{"error": "Unknown tool: fake"}' 1229 1230 with patch("tools.registry.registry", mock_registry): 1231 result = ctx.dispatch_tool("fake", {}) 1232 1233 assert '"error"' in result