Add Hermes lazy Agency router plugin (#614)

* Add Hermes lazy agency router plugin

* Document Hermes router specialist usage

scripts/build-hermes-plugin.py

@@ -0,0 +1,482 @@
+#!/usr/bin/env python3
+"""Build the Hermes lazy-router plugin for The Agency agents.
+
+The generated plugin exposes a small fixed tool surface to Hermes and keeps the
+large agent roster in an on-disk JSON data file. That avoids using
+skills.external_dirs, which advertises every Agency agent in Hermes' initial
+skill catalog.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import shutil
+import textwrap
+from pathlib import Path
+
+AGENT_DIRS = [
+    "academic",
+    "design",
+    "engineering",
+    "finance",
+    "game-development",
+    "gis",
+    "marketing",
+    "paid-media",
+    "product",
+    "project-management",
+    "sales",
+    "security",
+    "spatial-computing",
+    "specialized",
+    "support",
+    "testing",
+]
+
+PLUGIN_NAME = "agency-agents-router"
+
+
+def slugify(value: str) -> str:
+    value = value.lower()
+    value = re.sub(r"[^a-z0-9]+", "-", value)
+    return value.strip("-")
+
+
+def parse_agent(path: Path, repo_root: Path) -> dict[str, str] | None:
+    text = path.read_text(encoding="utf-8")
+    if not text.startswith("---\n"):
+        return None
+    parts = text.split("---\n", 2)
+    if len(parts) < 3:
+        return None
+    frontmatter = parts[1]
+    body = parts[2].lstrip("\n")
+    fields: dict[str, str] = {}
+    for line in frontmatter.splitlines():
+        if ":" not in line or line.startswith((" ", "\t")):
+            continue
+        key, value = line.split(":", 1)
+        fields[key.strip()] = value.strip().strip('"').strip("'")
+    name = fields.get("name", "").strip()
+    if not name:
+        return None
+    rel = path.relative_to(repo_root)
+    division = rel.parts[0]
+    return {
+        "slug": slugify(name),
+        "name": name,
+        "description": fields.get("description", "").strip(),
+        "division": division,
+        "color": fields.get("color", "").strip(),
+        "emoji": fields.get("emoji", "").strip(),
+        "vibe": fields.get("vibe", "").strip(),
+        "source_path": str(rel),
+        "body": body,
+    }
+
+
+def collect_agents(repo_root: Path) -> list[dict[str, str]]:
+    agents: list[dict[str, str]] = []
+    for dirname in AGENT_DIRS:
+        base = repo_root / dirname
+        if not base.is_dir():
+            continue
+        for path in sorted(base.rglob("*.md")):
+            parsed = parse_agent(path, repo_root)
+            if parsed:
+                agents.append(parsed)
+    agents.sort(key=lambda item: (item["division"], item["slug"]))
+    seen: set[str] = set()
+    duplicates: set[str] = set()
+    for agent in agents:
+        slug = agent["slug"]
+        if slug in seen:
+            duplicates.add(slug)
+        seen.add(slug)
+    if duplicates:
+        dupes = ", ".join(sorted(duplicates))
+        raise SystemExit(f"duplicate Hermes agent slugs: {dupes}")
+    return agents
+
+
+def plugin_yaml() -> str:
+    return textwrap.dedent(
+        f"""
+        name: {PLUGIN_NAME}
+        version: 1.0.0
+        description: Lazy search/load/delegate router for The Agency agent roster.
+        provides_tools:
+          - agency_agents_search
+          - agency_agents_inspect
+          - agency_agents_load
+          - agency_agents_delegate
+        """
+    ).lstrip()
+
+
+def init_py() -> str:
+    return r'''"""Hermes plugin: lazy router for The Agency agents."""
+from __future__ import annotations
+
+import json
+import math
+import re
+from pathlib import Path
+from typing import Any
+
+_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)
+
+
+def _load_agents() -> list[dict[str, Any]]:
+    global _AGENTS
+    if _AGENTS is None:
+        _AGENTS = json.loads(_DATA_PATH.read_text(encoding="utf-8"))
+    return _AGENTS
+
+
+def _tokens(text: str) -> set[str]:
+    return {token.lower() for token in _WORD_RE.findall(text or "")}
+
+
+def _agent_lookup(identifier: str) -> dict[str, Any] | None:
+    needle = (identifier or "").strip().lower()
+    if not needle:
+        return None
+    slug = re.sub(r"[^a-z0-9]+", "-", needle).strip("-")
+    for agent in _load_agents():
+        if agent["slug"] == slug or agent["name"].lower() == needle:
+            return agent
+    return None
+
+
+def _score(agent: dict[str, Any], query_tokens: set[str], query_text: str) -> float:
+    haystack_fields = [
+        agent.get("name", ""),
+        agent.get("description", ""),
+        agent.get("division", ""),
+        agent.get("vibe", ""),
+        agent.get("body", "")[:8000],
+    ]
+    haystack_text = "\n".join(haystack_fields).lower()
+    haystack_tokens = _tokens(haystack_text)
+    overlap = query_tokens & haystack_tokens
+    score = float(len(overlap))
+    if query_text and query_text in haystack_text:
+        score += 5.0
+    name = agent.get("name", "").lower()
+    description = agent.get("description", "").lower()
+    for token in query_tokens:
+        if token in name:
+            score += 3.0
+        if token in description:
+            score += 1.5
+    if score == 0.0:
+        return 0.0
+    # Slightly prefer focused descriptions over huge bodies when scores tie.
+    return score + (1.0 / math.sqrt(max(len(haystack_tokens), 1)))
+
+
+def _summary(agent: dict[str, Any], score: float | None = None) -> dict[str, Any]:
+    item = {
+        "slug": agent["slug"],
+        "name": agent["name"],
+        "division": agent["division"],
+        "description": agent.get("description", ""),
+        "vibe": agent.get("vibe", ""),
+        "source_path": agent.get("source_path", ""),
+    }
+    if score is not None:
+        item["score"] = round(score, 3)
+    return item
+
+
+def _specialist_prompt(agent: dict[str, Any], task: str = "") -> str:
+    task_block = f"\n\n## User task\n{task.strip()}\n" if task and task.strip() else ""
+    return (
+        f"Use the following Agency specialist context for this turn. "
+        f"Adopt the specialist's relevant standards and checklists, but obey the "
+        f"user's current request and higher-priority system/developer instructions.\n\n"
+        f"# {agent['name']} ({agent['slug']})\n\n"
+        f"Division: {agent.get('division', '')}\n"
+        f"Description: {agent.get('description', '')}\n"
+        f"Source: {agent.get('source_path', '')}\n"
+        f"{task_block}\n\n"
+        f"## Specialist instructions\n{agent.get('body', '')}"
+    )
+
+
+def _json(payload: dict[str, Any]) -> str:
+    return json.dumps(payload, ensure_ascii=False, indent=2)
+
+
+SEARCH_DESCRIPTION = (
+    "Search The Agency's on-disk specialist agent roster without loading all "
+    "agents into the prompt. Use this when the user asks for an Agency/Data "
+    "Swami specialist, role, discipline, or wants help choosing the right agent."
+)
+SEARCH_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "query": {"type": "string", "description": "Natural-language search query."},
+        "division": {"type": "string", "description": "Optional division filter, e.g. engineering, marketing, testing."},
+        "limit": {"type": "integer", "description": "Maximum results, default 8, max 25."},
+    },
+    "required": ["query"],
+}
+
+READ_DESCRIPTION = (
+    "Read one Agency specialist by slug or name. Returns metadata by default "
+    "and includes the full specialist instructions only when include_body is true."
+)
+READ_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "agent": {"type": "string", "description": "Agent slug or exact display name."},
+        "include_body": {"type": "boolean", "description": "Include full specialist instructions."},
+    },
+    "required": ["agent"],
+}
+
+PROMPT_DESCRIPTION = (
+    "Load a selected Agency specialist as a prompt block for the current task. "
+    "Use after agency_agents_search when you need one specialist's full context."
+)
+PROMPT_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "agent": {"type": "string", "description": "Agent slug or exact display name."},
+        "task": {"type": "string", "description": "The user's task to pair with the specialist context."},
+    },
+    "required": ["agent"],
+}
+
+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."
+)
+DELEGATE_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "agent": {"type": "string", "description": "Agent slug or exact display name."},
+        "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": ["agent", "task"],
+}
+
+
+def register(ctx):
+    def search(args: dict[str, Any], **kwargs) -> str:
+        del kwargs
+        query = str(args.get("query", "")).strip()
+        if not query:
+            return _json({"success": False, "error": "query is required"})
+        division = str(args.get("division", "")).strip().lower()
+        try:
+            limit = min(max(int(args.get("limit", 8)), 1), 25)
+        except Exception:
+            limit = 8
+        q_tokens = _tokens(query)
+        q_text = query.lower()
+        matches: list[tuple[float, dict[str, Any]]] = []
+        for agent in _load_agents():
+            if division and agent.get("division", "").lower() != division:
+                continue
+            score = _score(agent, q_tokens, q_text)
+            if score > 0:
+                matches.append((score, agent))
+        matches.sort(key=lambda item: (-item[0], item[1]["division"], item[1]["slug"]))
+        return _json({
+            "success": True,
+            "query": query,
+            "count": len(matches),
+            "results": [_summary(agent, score) for score, agent in matches[:limit]],
+        })
+
+    def read(args: dict[str, Any], **kwargs) -> str:
+        del kwargs
+        agent = _agent_lookup(str(args.get("agent", "")))
+        if not agent:
+            return _json({"success": False, "error": "agent not found", "agent": args.get("agent")})
+        payload = {"success": True, "agent": _summary(agent)}
+        if bool(args.get("include_body", False)):
+            payload["body"] = agent.get("body", "")
+        return _json(payload)
+
+    def prompt(args: dict[str, Any], **kwargs) -> str:
+        del kwargs
+        agent = _agent_lookup(str(args.get("agent", "")))
+        if not agent:
+            return _json({"success": False, "error": "agent not found", "agent": args.get("agent")})
+        return _json({
+            "success": True,
+            "agent": _summary(agent),
+            "prompt": _specialist_prompt(agent, str(args.get("task", ""))),
+        })
+
+    def delegate(args: dict[str, Any], **kwargs) -> str:
+        del kwargs
+        agent = _agent_lookup(str(args.get("agent", "")))
+        task = str(args.get("task", "")).strip()
+        if not agent:
+            return _json({"success": False, "error": "agent not found", "agent": args.get("agent")})
+        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]
+        try:
+            result = ctx.dispatch_tool("delegate_task", delegate_args)
+            return _json({"success": True, "agent": _summary(agent), "delegated": True, "result": result})
+        except Exception as exc:  # pragma: no cover - depends on Hermes runtime
+            return _json({
+                "success": True,
+                "agent": _summary(agent),
+                "delegated": False,
+                "warning": f"delegate_task unavailable: {exc}",
+                "prompt": composed,
+            })
+
+    ctx.register_tool(
+        name="agency_agents_search",
+        toolset="agency_agents",
+        schema=SEARCH_SCHEMA,
+        handler=search,
+        description=SEARCH_DESCRIPTION,
+    )
+    ctx.register_tool(
+        name="agency_agents_inspect",
+        toolset="agency_agents",
+        schema=READ_SCHEMA,
+        handler=read,
+        description=READ_DESCRIPTION,
+    )
+    ctx.register_tool(
+        name="agency_agents_load",
+        toolset="agency_agents",
+        schema=PROMPT_SCHEMA,
+        handler=prompt,
+        description=PROMPT_DESCRIPTION,
+    )
+    ctx.register_tool(
+        name="agency_agents_delegate",
+        toolset="agency_agents",
+        schema=DELEGATE_SCHEMA,
+        handler=delegate,
+        description=DELEGATE_DESCRIPTION,
+    )
+'''
+
+
+def readme(agent_count: int) -> str:
+    return textwrap.dedent(
+        f"""
+        # Hermes Agency Agents Router Plugin
+
+        Generated by `scripts/convert.sh --tool hermes`.
+
+        This integration installs one Hermes plugin named `{PLUGIN_NAME}` instead
+        of adding 232+ generated skills to `skills.external_dirs`. Hermes sees a
+        small fixed tool surface at startup, while the complete Agency roster is
+        stored on disk in `data/agents.json` and searched/loaded lazily.
+
+        Generated agent count: {agent_count}
+
+        ## Tools exposed to Hermes
+
+        - `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.
+
+        ## Specialist usage instruction for Hermes
+
+        When a Hermes project needs Agency specialists, explicitly ask Hermes to use
+        the `{PLUGIN_NAME}` plugin/router and load only the specialists needed for
+        the current phase. Do not ask Hermes to install or preload the full Agency
+        roster as skills.
+
+        Recommended project instruction:
+
+        ```text
+        Use the agency-agents-router plugin. Search the Agency roster for the right
+        specialists, then load or delegate only the specific agents needed for each
+        part of the project. For multi-discipline projects, use multiple selected
+        specialists across the project, but keep routing lazy: do not preload the
+        full Agency roster and do not add agency-agents to skills.external_dirs.
+        ```
+
+        Example:
+
+        ```text
+        For this Data Swami build, use the agency-agents-router plugin to pick
+        relevant Agency specialists. Search first, then delegate to selected agents
+        such as frontend, backend, UX, QA, data engineering, and product strategy as
+        needed. Load/delegate each specialist on demand rather than loading all
+        Agency agents at startup.
+        ```
+
+        ## Install
+
+        ```bash
+        ./scripts/convert.sh --tool hermes
+        ./scripts/install.sh --tool hermes
+        ```
+
+        The installer copies the generated plugin to:
+
+        ```text
+        ${{HERMES_HOME:-~/.hermes}}/plugins/{PLUGIN_NAME}
+        ```
+
+        It then enables `{PLUGIN_NAME}` under `plugins.enabled` in the Hermes
+        config. It does **not** write to `skills.external_dirs`.
+        """
+    ).lstrip()
+
+
+def build(repo_root: Path, out_dir: Path) -> int:
+    agents = collect_agents(repo_root)
+    plugin_dir = out_dir / PLUGIN_NAME
+    if plugin_dir.exists():
+        shutil.rmtree(plugin_dir)
+    (plugin_dir / "data").mkdir(parents=True, exist_ok=True)
+    (plugin_dir / "plugin.yaml").write_text(plugin_yaml(), encoding="utf-8")
+    (plugin_dir / "__init__.py").write_text(init_py(), encoding="utf-8")
+    (plugin_dir / "data" / "agents.json").write_text(
+        json.dumps(agents, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    (out_dir / "README.md").write_text(readme(len(agents)), encoding="utf-8")
+    return len(agents)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument("--out", type=Path, default=None, help="Output directory, default integrations/hermes")
+    args = parser.parse_args()
+    repo_root = args.repo_root.resolve()
+    out_dir = (args.out or (repo_root / "integrations" / "hermes")).resolve()
+    out_dir.mkdir(parents=True, exist_ok=True)
+    count = build(repo_root, out_dir)
+    print(count)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())