diff --git a/integrations/hermes/README.md b/integrations/hermes/README.md index 183bbb02..9c8b4c15 100644 --- a/integrations/hermes/README.md +++ b/integrations/hermes/README.md @@ -14,7 +14,7 @@ Generated agent count: 273 - `agency_agents_search` — find matching specialists by query/division. - `agency_agents_inspect` — inspect one specialist's metadata or full body. - `agency_agents_load` — compose one specialist prompt for the current task. -- `agency_agents_delegate` — delegate through Hermes `delegate_task` when available. +- `agency_agents_delegate` — delegate through Hermes' public subagent lifecycle. Each tool is registered with Hermes' complete function-tool schema, including its name, description, and JSON `parameters`. The available arguments are: @@ -24,7 +24,7 @@ its name, description, and JSON `parameters`. The available arguments are: | `agency_agents_search` | `query` (required), optional `division` and `limit` | | `agency_agents_inspect` | `agent` or `slug`, optional `include_body` | | `agency_agents_load` | `agent` or `slug`, optional `task` | -| `agency_agents_delegate` | `agent` or `slug`, `task` (required), optional `toolsets` | +| `agency_agents_delegate` | `agent` or `slug`, `task` (required) | A normal flow is: search by capability, take a returned `slug`, then inspect, load, or delegate to that specialist. You can ask Hermes to do this in natural diff --git a/scripts/build-hermes-plugin.py b/scripts/build-hermes-plugin.py index edd62a26..caffc56b 100644 --- a/scripts/build-hermes-plugin.py +++ b/scripts/build-hermes-plugin.py @@ -120,6 +120,12 @@ _DATA_PATH = Path(__file__).parent / "data" / "agents.json" _AGENTS: list[dict[str, Any]] | None = None _WORD_RE = re.compile(r"[a-z0-9][a-z0-9+.#_-]*", re.I) +_MAX_LIFECYCLE_CONTEXT_CHARS = 32_000 +_DELEGATION_WAIT_SECONDS = 330 +_CANCELLATION_WAIT_SECONDS = 30 +_TRUNCATION_MARKER = ( + "\n\n[Specialist instructions truncated to fit the Hermes lifecycle context limit.]" +) def _load_agents() -> list[dict[str, Any]]: @@ -215,6 +221,14 @@ def _specialist_prompt(agent: dict[str, Any], task: str = "") -> str: ) +def _lifecycle_context(agent: dict[str, Any]) -> str: + context = _specialist_prompt(agent) + if len(context) <= _MAX_LIFECYCLE_CONTEXT_CHARS: + return context + keep = _MAX_LIFECYCLE_CONTEXT_CHARS - len(_TRUNCATION_MARKER) + return context[:keep] + _TRUNCATION_MARKER + + def _json(payload: dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, indent=2) @@ -276,8 +290,8 @@ PROMPT_SCHEMA = { DELEGATE_DESCRIPTION = ( "Delegate a task to one selected Agency specialist through Hermes' " - "delegate_task tool when available. Falls back to returning the composed " - "specialist prompt if delegation is unavailable." + "public subagent lifecycle. Falls back to returning the composed specialist " + "prompt if delegation is unavailable." ) DELEGATE_SCHEMA = { "name": "agency_agents_delegate", @@ -288,11 +302,6 @@ DELEGATE_SCHEMA = { "agent": {"type": "string", "description": "Agent slug or exact display name."}, "slug": {"type": "string", "description": "Alias for agent. Pass the slug from agency_agents_search results."}, "task": {"type": "string", "description": "Concrete task for the specialist."}, - "toolsets": { - "type": "array", - "items": {"type": "string"}, - "description": "Optional Hermes toolsets for the delegated worker, e.g. ['terminal','file'].", - }, }, "required": ["task"], }, @@ -359,24 +368,85 @@ def register(ctx): return _json(_not_found(identifier)) if not task: return _json({"success": False, "error": "task is required"}) - composed = _specialist_prompt(agent, task) - delegate_args: dict[str, Any] = { - "goal": task, - "context": composed, - } - toolsets = args.get("toolsets") - if isinstance(toolsets, list) and toolsets: - delegate_args["toolsets"] = [str(item) for item in toolsets] + fallback_prompt = _specialist_prompt(agent, task) + handle = None try: - result = ctx.dispatch_tool("delegate_task", delegate_args) - return _json({"success": True, "agent": _summary(agent), "delegated": True, "result": result}) + from agent.subagent_lifecycle import SubagentLaunchRequest + + lifecycle = ctx.subagent_lifecycle + handle = lifecycle.launch(SubagentLaunchRequest( + goal=task, + context=_lifecycle_context(agent), + )) + terminal = lifecycle.wait( + handle, timeout_seconds=_DELEGATION_WAIT_SECONDS + ) + if terminal.timed_out: + try: + lifecycle.cancel( + handle, + reason="Agency delegation exceeded the plugin wait limit.", + ) + terminal = lifecycle.wait( + handle, timeout_seconds=_CANCELLATION_WAIT_SECONDS + ) + except Exception as exc: + return _json({ + "success": True, + "agent": _summary(agent), + "delegated": True, + "pending": True, + "subagent_id": handle.subagent_id, + "warning": f"subagent cancellation could not be confirmed: {exc}", + }) + if not terminal.completed: + return _json({ + "success": True, + "agent": _summary(agent), + "delegated": True, + "pending": True, + "subagent_id": handle.subagent_id, + "state": terminal.state.value, + "warning": "subagent cancellation was requested but is not terminal", + }) + result = lifecycle.result(handle) + if not result.ready or result.terminal_state.value != "SUCCEEDED": + detail = ( + result.error_message + or result.error_classification + or result.terminal_state.value + ) + return _json({ + "success": True, + "agent": _summary(agent), + "delegated": False, + "warning": f"subagent delegation failed: {detail}", + "prompt": fallback_prompt, + }) + return _json({ + "success": True, + "agent": _summary(agent), + "delegated": True, + "subagent_id": handle.subagent_id, + "result": result.summary, + "structured_result": result.structured_payload, + }) except Exception as exc: # pragma: no cover - depends on Hermes runtime + if handle is not None: + return _json({ + "success": True, + "agent": _summary(agent), + "delegated": True, + "pending": True, + "subagent_id": handle.subagent_id, + "warning": f"subagent state could not be confirmed: {exc}", + }) return _json({ "success": True, "agent": _summary(agent), "delegated": False, - "warning": f"delegate_task unavailable: {exc}", - "prompt": composed, + "warning": f"subagent delegation unavailable: {exc}", + "prompt": fallback_prompt, }) ctx.register_tool( @@ -429,7 +499,7 @@ def readme(agent_count: int) -> str: - `agency_agents_search` — find matching specialists by query/division. - `agency_agents_inspect` — inspect one specialist's metadata or full body. - `agency_agents_load` — compose one specialist prompt for the current task. - - `agency_agents_delegate` — delegate through Hermes `delegate_task` when available. + - `agency_agents_delegate` — delegate through Hermes' public subagent lifecycle. Each tool is registered with Hermes' complete function-tool schema, including its name, description, and JSON `parameters`. The available arguments are: @@ -439,7 +509,7 @@ def readme(agent_count: int) -> str: | `agency_agents_search` | `query` (required), optional `division` and `limit` | | `agency_agents_inspect` | `agent` or `slug`, optional `include_body` | | `agency_agents_load` | `agent` or `slug`, optional `task` | - | `agency_agents_delegate` | `agent` or `slug`, `task` (required), optional `toolsets` | + | `agency_agents_delegate` | `agent` or `slug`, `task` (required) | A normal flow is: search by capability, take a returned `slug`, then inspect, load, or delegate to that specialist. You can ask Hermes to do this in natural diff --git a/scripts/test-hermes-plugin.py b/scripts/test-hermes-plugin.py new file mode 100644 index 00000000..a6853dd4 --- /dev/null +++ b/scripts/test-hermes-plugin.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Behavior checks for the generated Hermes Agency router plugin.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import types +import unittest +from enum import Enum +from pathlib import Path +from types import SimpleNamespace + +ROOT = Path(__file__).resolve().parents[1] +PLUGIN = ROOT / "integrations" / "hermes" / "agency-agents-router" / "__init__.py" + + +class State(str, Enum): + RUNNING = "RUNNING" + CANCEL_REQUESTED = "CANCEL_REQUESTED" + SUCCEEDED = "SUCCEEDED" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + + +class FakeRequest: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + +class FakeLifecycle: + def __init__(self, *, waits=None, result=None, launch_error=None): + self.waits = list(waits or [terminal(State.SUCCEEDED, completed=True)]) + self.result_value = result or child_result(State.SUCCEEDED, summary="ROUTER_OK") + self.launch_error = launch_error + self.request = None + self.wait_timeouts = [] + self.cancelled = False + + def launch(self, request): + if self.launch_error: + raise self.launch_error + self.request = request + return SimpleNamespace(subagent_id="child-1") + + def wait(self, handle, *, timeout_seconds=None): + del handle + self.wait_timeouts.append(timeout_seconds) + return self.waits.pop(0) + + def cancel(self, handle, *, reason): + del handle, reason + self.cancelled = True + return SimpleNamespace(accepted=True) + + def result(self, handle): + del handle + return self.result_value + + +class FakeContext: + def __init__(self, lifecycle): + self.subagent_lifecycle = lifecycle + self.tools = {} + + def register_tool(self, *, name, schema, handler, **kwargs): + del kwargs + self.tools[name] = (schema, handler) + + +def terminal(state, *, completed=False, timed_out=False): + return SimpleNamespace(state=state, completed=completed, timed_out=timed_out) + + +def child_result(state, *, ready=True, summary=None, error=None): + return SimpleNamespace( + ready=ready, + terminal_state=state, + summary=summary, + structured_payload={"ok": True} if state == State.SUCCEEDED else None, + error_message=error, + error_classification=None, + ) + + +def load_plugin(lifecycle): + agent_module = types.ModuleType("agent") + lifecycle_module = types.ModuleType("agent.subagent_lifecycle") + setattr(lifecycle_module, "SubagentLaunchRequest", FakeRequest) + sys.modules["agent"] = agent_module + sys.modules["agent.subagent_lifecycle"] = lifecycle_module + + spec = importlib.util.spec_from_file_location("agency_router_under_test", PLUGIN) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + context = FakeContext(lifecycle) + module.register(context) + return module, context + + +class DelegateBehaviorTests(unittest.TestCase): + def invoke(self, lifecycle, slug="ux-architect"): + module, context = load_plugin(lifecycle) + schema, handler = context.tools["agency_agents_delegate"] + payload = json.loads(handler({"slug": slug, "task": "Return ROUTER_OK"})) + return module, schema, payload + + def test_success_returns_child_result_without_toolsets_option(self): + lifecycle = FakeLifecycle() + _, schema, payload = self.invoke(lifecycle) + self.assertTrue(payload["delegated"]) + self.assertEqual(payload["result"], "ROUTER_OK") + self.assertNotIn("toolsets", schema["parameters"]["properties"]) + self.assertEqual(lifecycle.wait_timeouts, [330]) + + def test_failed_child_returns_safe_fallback(self): + lifecycle = FakeLifecycle( + result=child_result(State.FAILED, error="child failed") + ) + _, _, payload = self.invoke(lifecycle) + self.assertFalse(payload["delegated"]) + self.assertIn("Return ROUTER_OK", payload["prompt"]) + + def test_launch_exception_returns_safe_fallback(self): + lifecycle = FakeLifecycle(launch_error=RuntimeError("launch failed")) + _, _, payload = self.invoke(lifecycle) + self.assertFalse(payload["delegated"]) + self.assertIn("launch failed", payload["warning"]) + + def test_unconfirmed_cancellation_is_truthfully_pending_without_fallback(self): + lifecycle = FakeLifecycle(waits=[ + terminal(State.RUNNING, timed_out=True), + terminal(State.CANCEL_REQUESTED), + ]) + _, _, payload = self.invoke(lifecycle) + self.assertTrue(payload["delegated"]) + self.assertTrue(payload["pending"]) + self.assertNotIn("prompt", payload) + self.assertTrue(lifecycle.cancelled) + self.assertEqual(lifecycle.wait_timeouts, [330, 30]) + + def test_confirmed_cancellation_returns_fallback(self): + lifecycle = FakeLifecycle( + waits=[ + terminal(State.RUNNING, timed_out=True), + terminal(State.CANCELLED, completed=True), + ], + result=child_result(State.CANCELLED, error="cancelled"), + ) + _, _, payload = self.invoke(lifecycle) + self.assertFalse(payload["delegated"]) + self.assertIn("Return ROUTER_OK", payload["prompt"]) + + def test_large_specialist_context_is_marked_and_bounded(self): + lifecycle = FakeLifecycle() + module, _, payload = self.invoke( + lifecycle, slug="healthcare-marketing-compliance-specialist" + ) + self.assertTrue(payload["delegated"]) + request = lifecycle.request + self.assertIsNotNone(request) + assert request is not None + self.assertEqual(len(request.context), 32_000) + self.assertTrue(request.context.endswith(module._TRUNCATION_MARKER)) + + +if __name__ == "__main__": + unittest.main()