diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py index 849ac896..72228cb5 100644 --- a/src/bmad_loop/adapters/generic.py +++ b/src/bmad_loop/adapters/generic.py @@ -283,7 +283,7 @@ def build_command(self, spec: SessionSpec) -> str: # --------------------------------------------------------------- adapter - def start_session(self, spec: SessionSpec) -> SessionHandle: + def _prepare_session(self, spec: SessionSpec) -> None: task_dir = self.tasks_dir / spec.task_id task_dir.mkdir(parents=True, exist_ok=True) (task_dir / "prompt.txt").write_text(spec.prompt + "\n", encoding="utf-8") @@ -291,6 +291,7 @@ def start_session(self, spec: SessionSpec) -> SessionHandle: # so a session that writes nothing can't be read as a stale completion. (task_dir / "result.json").unlink(missing_ok=True) + def _launch_session(self, spec: SessionSpec, command: str) -> SessionHandle: self._ensure_session(spec.cwd) # Stamped before launch: hook events carry wall-clock ns, and # wait_for_completion ignores anything older than this floor so a reused @@ -300,8 +301,8 @@ def start_session(self, spec: SessionSpec) -> SessionHandle: self.session_name, spec.task_id[-40:], spec.cwd, - {**self.profile.env, **spec.env}, - self.build_command(spec), + self.interactive_env(spec), + command, ) log_file = self.logs_dir / f"{spec.task_id}.log" # pipe_pane tolerates the window having already died (a CLI that crashes on @@ -310,6 +311,10 @@ def start_session(self, spec: SessionSpec) -> SessionHandle: self.mux.pipe_pane(window_id, log_file) return SessionHandle(task_id=spec.task_id, native_id=window_id, launched_ns=launched_ns) + def start_session(self, spec: SessionSpec) -> SessionHandle: + self._prepare_session(spec) + return self._launch_session(spec, self.build_command(spec)) + def wait_for_completion(self, handle: SessionHandle, spec: SessionSpec) -> SessionResult: deadline = time.monotonic() + spec.timeout_s # Wall-clock co-bound (#157): a host suspend freezes time.monotonic(), diff --git a/src/bmad_loop/adapters/hermes.py b/src/bmad_loop/adapters/hermes.py new file mode 100644 index 00000000..30c04cbc --- /dev/null +++ b/src/bmad_loop/adapters/hermes.py @@ -0,0 +1,53 @@ +"""Interactive Hermes Agent adapter for attachable BMAD Loop tmux sessions.""" + +from __future__ import annotations + +import time +from typing import Any +from ..bmadconfig import ProjectPaths +from .base import SessionHandle, SessionSpec +from .generic import GenericAdapter, _DevSynthesisMixin + + +class HermesStartupError(RuntimeError): + """Hermes exited before the adapter could deliver the initial prompt.""" + + +class HermesAdapter(GenericAdapter): + """Launch normal Hermes CLI sessions and inject prompts after startup.""" + + STARTUP_GRACE_S = 2.0 + + def interactive_argv(self, spec: SessionSpec) -> list[str]: + extra = self.extra_args + if extra is None: + extra = self.profile.bypass_args + argv = [self.binary, *self.profile.launch_args, *extra] + if spec.model: + argv += [self.profile.model_flag, spec.model] + return argv + + def interactive_env(self, spec: SessionSpec) -> dict[str, str]: + env = super().interactive_env(spec) + # Hermes discovers project-local skills through this explicit directory. + # Respect a caller override for custom shared skill layouts. + env.setdefault("HERMES_PROJECT_SKILLS", str((spec.cwd / self.profile.skill_tree).resolve())) + return env + + def start_session(self, spec: SessionSpec) -> SessionHandle: + self._prepare_session(spec) + handle = self._launch_session(spec, self.build_command(spec)) + time.sleep(self.STARTUP_GRACE_S) + if not self._window_alive(handle): + raise HermesStartupError(f"{spec.task_id}: pane exited during Hermes startup") + self.send_text(handle, self.profile.render_prompt(spec.prompt)) + return handle + + +class HermesDevAdapter(_DevSynthesisMixin, HermesAdapter): + """Hermes dev/review adapter that synthesizes bmad-dev-auto results from specs.""" + + def __init__(self, *args: Any, paths: ProjectPaths, **kwargs: Any): + super().__init__(*args, **kwargs) + self.paths = paths + self._configure_dev_knobs() diff --git a/src/bmad_loop/adapters/profile.py b/src/bmad_loop/adapters/profile.py index e7d242d0..190da8e2 100644 --- a/src/bmad_loop/adapters/profile.py +++ b/src/bmad_loop/adapters/profile.py @@ -27,12 +27,15 @@ "gemini-settings-json", "copilot-settings-json", "antigravity-hooks-json", + "hermes-config-yaml", # hookless: the adapter observes completion itself (HTTP/SSE transport) — # no hook config is ever written, so config_path/events must stay empty. "none", } CANONICAL_EVENTS = {"SessionStart", "Stop", "SessionEnd", "PreCompact"} USER_PROFILES_REL = Path(".bmad-loop") / "profiles" +ADAPTER_KINDS = {"generic", "hermes"} +HOOK_SCOPES = {"project", "user"} # legacy adapter names from older policy.toml files, plus friendly short names ALIASES = {"claude-code-tmux": "claude", "opencode": "opencode-http"} @@ -54,6 +57,8 @@ class CLIProfile: name: str binary: str hooks: HookSpec + adapter: str = "generic" + hook_scope: str = "project" # project-relative tree this CLI reads skills from, e.g. ".claude/skills" # (claude) or ".agents/skills" (codex/gemini); `bmad-loop init` installs the # bundled bmad-loop-* skills here. @@ -117,6 +122,17 @@ def fail(msg: str) -> ProfileError: if not name or not binary: raise fail("'name' and 'binary' are required") + adapter = str(doc.get("adapter", "generic")) + if adapter not in ADAPTER_KINDS: + raise fail(f"adapter must be one of {sorted(ADAPTER_KINDS)}: got {adapter!r}") + hook_scope = str(doc.get("hook_scope", "project")) + if hook_scope not in HOOK_SCOPES: + raise fail(f"hook_scope must be one of {sorted(HOOK_SCOPES)}: got {hook_scope!r}") + if adapter == "hermes" and hook_scope != "user": + raise fail('adapter = "hermes" requires hook_scope = "user"') + if hook_scope == "user" and adapter != "hermes": + raise fail('hook_scope = "user" requires adapter = "hermes"') + hooks_d = doc.get("hooks") if not isinstance(hooks_d, dict): raise fail("missing [hooks] table") @@ -132,7 +148,13 @@ def fail(msg: str) -> ProfileError: events: dict[str, str] = {} else: config_path = str(hooks_d.get("config_path", "")) - if not config_path or is_absolute_path(config_path) or has_parent_ref(config_path): + if hook_scope == "user": + if dialect != "hermes-config-yaml" or config_path != "config.yaml": + raise fail( + 'user-scoped hooks require dialect = "hermes-config-yaml" ' + 'and config_path = "config.yaml"' + ) + elif not config_path or is_absolute_path(config_path) or has_parent_ref(config_path): raise fail("hooks.config_path must be a project-relative path") events_d = hooks_d.get("events") if not isinstance(events_d, dict) or not events_d: @@ -170,6 +192,8 @@ def fail(msg: str) -> ProfileError: name=name, binary=binary, hooks=HookSpec(dialect=dialect, config_path=config_path, events=events), + adapter=adapter, + hook_scope=hook_scope, skill_tree=skill_tree, prompt_template=str(doc.get("prompt_template", "{prompt}")), launch_args=tuple(str(a) for a in doc.get("launch_args", ())), diff --git a/src/bmad_loop/cli.py b/src/bmad_loop/cli.py index 7b1cab42..f91ab0cd 100644 --- a/src/bmad_loop/cli.py +++ b/src/bmad_loop/cli.py @@ -31,7 +31,9 @@ verify, ) from .adapters.base import CodingCLIAdapter +from .adapters.profile import CANONICAL_EVENTS from .checks import Finding, ValidationReport +from .hermes_hooks import relay_event # The --json document builders live in documents.py (the library-level projection # layer a non-CLI frontend imports). The schema constants are re-exported rather @@ -199,11 +201,20 @@ def _make_adapters(project: Path, run_dir: Path, policy) -> dict[str, CodingCLIA stop_without_result_nudges=cfg.stop_without_result_nudges, mux=mux, ) - by_cfg[key] = ( - GenericDevAdapter(**common, paths=paths) - if synthesizes - else GenericAdapter(**common) - ) + if profile.adapter == "hermes": + from .adapters.hermes import HermesAdapter, HermesDevAdapter + + by_cfg[key] = ( + HermesDevAdapter(**common, paths=paths) + if synthesizes + else HermesAdapter(**common) + ) + else: + by_cfg[key] = ( + GenericDevAdapter(**common, paths=paths) + if synthesizes + else GenericAdapter(**common) + ) adapters[role] = by_cfg[key] return adapters @@ -339,8 +350,6 @@ def _platform_preflight() -> list[Finding]: def cmd_validate(args: argparse.Namespace) -> int: - from .install import relay_registered - project = _project(args) report = ValidationReport() @@ -449,20 +458,12 @@ def cmd_validate(args: argparse.Namespace) -> int: {"profile": profile.name}, ) continue - hook_config = project / profile.hooks.config_path - hooks_ok = False - if hook_config.is_file(): - try: - parsed = json.loads(hook_config.read_text(encoding="utf-8")) - hooks_ok = isinstance(parsed, dict) and relay_registered( - parsed, profile.hooks.dialect, profile.hooks.events - ) - except json.JSONDecodeError: - report.fail( - "hooks.config-parse", - f"{hook_config} is not valid JSON", - {"profile": profile.name, "config_path": str(hook_config)}, - ) + hook_config = ( + install.hermes_config_path(os.environ) + if profile.hook_scope == "user" + else project / profile.hooks.config_path + ) + hooks_ok = install.hooks_registered(project, profile) if hooks_ok: report.ok( "hooks.registered", @@ -2254,6 +2255,16 @@ def cmd_init(args: argparse.Namespace) -> int: return install_into(project, clis=clis, skills=args.skills, force_skills=args.force_skills) +def cmd_relay(args: argparse.Namespace) -> int: + """Receive a Hermes post-turn hook payload and emit a canonical BMAD event.""" + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, ValueError): + payload = {} + relay_event(args.event, payload if isinstance(payload, dict) else {}, os.environ) + return 0 + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="bmad-loop", @@ -2275,7 +2286,7 @@ def add(name: str, func, help: str, *, aliases=()) -> argparse.ArgumentParser: "--cli", action="append", metavar="PROFILE", - help="CLI profile(s) to register hooks for (claude | codex | gemini | copilot | " + help="CLI profile(s) to register hooks for (claude | codex | gemini | copilot | hermes | " "antigravity | opencode-http (alias: opencode) | custom; " "repeatable; default: profiles referenced by .bmad-loop/policy.toml, or claude)", ) @@ -2290,6 +2301,8 @@ def add(name: str, func, help: str, *, aliases=()) -> argparse.ArgumentParser: action="store_true", help="overwrite bmad-loop-* skill dirs that already exist (default: skip them)", ) + relay_p = add("relay", cmd_relay, "write one canonical event from a coding-CLI hook payload") + relay_p.add_argument("event", choices=sorted(CANONICAL_EVENTS)) validate_p = add("validate", cmd_validate, "preflight checks; exit non-zero on failure") validate_p.add_argument( "--spec", @@ -2331,7 +2344,7 @@ def add(name: str, func, help: str, *, aliases=()) -> argparse.ArgumentParser: ) probe_p.add_argument( "cli", - help="CLI profile name (claude | codex | gemini | copilot | antigravity | custom; " + help="CLI profile name (claude | codex | gemini | copilot | hermes | antigravity | custom; " "opencode-http is HTTP-driven — nothing to probe)", ) probe_p.add_argument( diff --git a/src/bmad_loop/data/profiles/hermes.toml b/src/bmad_loop/data/profiles/hermes.toml new file mode 100644 index 00000000..ff3de9dc --- /dev/null +++ b/src/bmad_loop/data/profiles/hermes.toml @@ -0,0 +1,17 @@ +# Hermes Agent uses its normal interactive REPL in an attachable tmux pane. +# The initial BMAD prompt is injected after startup rather than passed as argv. +name = "hermes" +binary = "hermes" +adapter = "hermes" +hook_scope = "user" +prompt_template = "Use the ${skill} skill now: {args}" +launch_args = ["--cli", "--accept-hooks", "--yolo"] +model_flag = "--model" +usage_parser = "none" +skill_tree = ".agents/skills" +first_run_note = "run `hermes --cli` once and approve the bmad-loop relay hook; spawned sessions pass --accept-hooks" + +[hooks] +dialect = "hermes-config-yaml" +config_path = "config.yaml" +events = { post_llm_call = "Stop" } diff --git a/src/bmad_loop/hermes_hooks.py b/src/bmad_loop/hermes_hooks.py new file mode 100644 index 00000000..ea327eb9 --- /dev/null +++ b/src/bmad_loop/hermes_hooks.py @@ -0,0 +1,104 @@ +"""Hermes user-config hooks and the package-owned BMAD event relay.""" + +from __future__ import annotations + +import copy +import json +import os +import time +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +HERMES_RELAY_COMMAND = "bmad-loop relay Stop" +HERMES_RELAY_HOOK = {"command": HERMES_RELAY_COMMAND, "timeout": 10} + + +def hermes_config_path(environ: Mapping[str, str]) -> Path: + """Return the active Hermes config without reading credentials or config content.""" + home = environ.get("HERMES_HOME", "~/.hermes") + return Path(home).expanduser() / "config.yaml" + + +def _copy_hook_entries(config: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]: + updated = copy.deepcopy(config) + hooks = updated.get("hooks") + if hooks is None: + hooks = {} + updated["hooks"] = hooks + if not isinstance(hooks, dict): + raise ValueError("Hermes config hooks must be a mapping") + entries = hooks.get("post_llm_call") + if entries is None: + entries = [] + hooks["post_llm_call"] = entries + if not isinstance(entries, list) or not all(isinstance(entry, dict) for entry in entries): + raise ValueError("Hermes hooks.post_llm_call must be a list of mappings") + return updated, entries + + +def merge_hermes_stop_hook(config: dict[str, Any]) -> tuple[dict[str, Any], bool]: + """Add the managed relay once while preserving all unrelated Hermes hooks.""" + updated, entries = _copy_hook_entries(config) + if any(entry.get("command") == HERMES_RELAY_COMMAND for entry in entries): + return updated, False + entries.append(copy.deepcopy(HERMES_RELAY_HOOK)) + return updated, True + + +def remove_hermes_stop_hook(config: dict[str, Any]) -> tuple[dict[str, Any], bool]: + """Remove only the managed relay, pruning empty maps left by that removal.""" + updated = copy.deepcopy(config) + hooks = updated.get("hooks") + if not isinstance(hooks, dict): + return updated, False + entries = hooks.get("post_llm_call") + if not isinstance(entries, list): + return updated, False + kept = [entry for entry in entries if not (isinstance(entry, dict) and entry.get("command") == HERMES_RELAY_COMMAND)] + if len(kept) == len(entries): + return updated, False + if kept: + hooks["post_llm_call"] = kept + else: + hooks.pop("post_llm_call", None) + if not hooks: + updated.pop("hooks", None) + return updated, True + + +def _first_workspace(payload: Mapping[str, Any]) -> str | None: + paths = payload.get("workspacePaths") + if isinstance(paths, list) and paths and isinstance(paths[0], str): + return paths[0] + return None + + +def relay_event(event_name: str, payload: Mapping[str, Any], environ: Mapping[str, str]) -> bool: + """Atomically write a canonical BMAD event only from an active BMAD session.""" + run_dir = environ.get("BMAD_LOOP_RUN_DIR") + task_id = environ.get("BMAD_LOOP_TASK_ID") + if not run_dir or not task_id: + return False + + ts = time.time_ns() + event = { + "ts": ts, + "event": event_name, + "task_id": task_id, + "session_id": ( + payload.get("session_id") + or payload.get("conversation_id") + or payload.get("sessionId") + or payload.get("conversationId") + ), + "transcript_path": payload.get("transcript_path") or payload.get("transcriptPath"), + "cwd": payload.get("cwd") or _first_workspace(payload), + } + events_dir = Path(run_dir) / "events" + events_dir.mkdir(parents=True, exist_ok=True) + final = events_dir / f"{ts}-{task_id}-{event_name}.json" + temporary = final.with_suffix(".json.tmp") + temporary.write_text(json.dumps(event), encoding="utf-8") + os.replace(temporary, final) + return True diff --git a/src/bmad_loop/install.py b/src/bmad_loop/install.py index ccb5b24c..0a53b015 100644 --- a/src/bmad_loop/install.py +++ b/src/bmad_loop/install.py @@ -17,14 +17,18 @@ from __future__ import annotations import json +import os import shutil import subprocess from collections.abc import Iterable, Sequence from importlib import resources from pathlib import Path +import yaml + from .adapters.profile import ALIASES, CLIProfile, ProfileError, load_profiles from .checks import Finding +from .hermes_hooks import HERMES_RELAY_COMMAND, hermes_config_path, merge_hermes_stop_hook from .policy import POLICY_TEMPLATE from .process_host import get_process_host @@ -259,9 +263,36 @@ def _managed_hook_in_handlers(handlers) -> bool: def relay_registered(config: dict, dialect: str, events: Iterable[str]) -> bool: """True if the bmad-loop relay is registered for any of `events`.""" container = hook_event_container(config, dialect) + if dialect == "hermes-config-yaml": + return any( + any( + isinstance(entry, dict) and entry.get("command") == HERMES_RELAY_COMMAND + for entry in container.get(event, []) + ) + for event in events + ) return any(_relay_in_handlers(container.get(event, [])) for event in events) +def hooks_registered(project: Path, profile: CLIProfile) -> bool: + """Read the profile's hook config and report whether its managed relay exists.""" + if profile.hook_scope == "user": + config_path = hermes_config_path(os.environ) + try: + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return False + else: + config_path = project / profile.hooks.config_path + try: + config = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return False + return isinstance(config, dict) and relay_registered( + config, profile.hooks.dialect, profile.hooks.events + ) + + def merge_hooks(config: dict, registrations: dict[str, str], dialect: str) -> tuple[dict, bool]: """Add relay registrations (native event -> command) to a hook config dict.""" changed = False @@ -384,6 +415,35 @@ def _register_hooks(project: Path, profile: CLIProfile) -> int: return 0 +def _register_user_scoped_hooks(profile: CLIProfile) -> int: + """Register Hermes's managed relay without disturbing its normal config.""" + config_path = hermes_config_path(os.environ) + config: dict = {} + if config_path.is_file(): + try: + loaded = yaml.safe_load(config_path.read_text(encoding="utf-8")) + except yaml.YAMLError: + print(f"FAIL: {config_path} is not valid YAML; fix it and re-run init") + return 1 + if loaded is not None: + if not isinstance(loaded, dict): + print(f"FAIL: {config_path} must contain a YAML mapping; fix it and re-run init") + return 1 + config = loaded + try: + config, changed = merge_hermes_stop_hook(config) + except ValueError as exc: + print(f"FAIL: {config_path}: {exc}") + return 1 + if changed: + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8") + print(f" hooks registered ({profile.name}): {config_path}") + else: + print(f" hooks already registered ({profile.name})") + return 0 + + def _copy_traversable(src, dst: Path) -> None: """Recursively copy a packaged resource tree to a filesystem path. @@ -582,10 +642,11 @@ def provision_worktree( continue _copy_traversable(src, dst) - # per-CLI signal-hook registration, baked to the main repo's relay (absolute). - # Hookless profiles (HTTP/SSE transport) have no config to merge. + # Per-CLI signal-hook registration, baked to the main repo's relay (absolute). + # User-scoped profiles (Hermes) inherit their already-registered HERMES_HOME + # hook and must never treat a worktree's config.yaml as their own config. for profile in profiles: - if profile.hookless: + if profile.hookless or profile.hook_scope != "project": continue config_path = worktree / profile.hooks.config_path config_path.parent.mkdir(parents=True, exist_ok=True) @@ -609,9 +670,14 @@ def provision_worktree( # configs) from the unit's `git add -A`, in case a project doesn't gitignore # its tool dirs. patterns = {f"/{p.skill_tree}" for p in profiles} - # hookless profiles have no config_path — and their empty string would render - # as the pattern "/", git-excluding the entire worktree. - patterns |= {f"/{p.hooks.config_path}" for p in profiles if not p.hookless} + # Hookless profiles have no config_path — and their empty string would render + # as the pattern "/", git-excluding the entire worktree. User-scoped profiles + # likewise own no worktree config and must not exclude a project's config.yaml. + patterns |= { + f"/{p.hooks.config_path}" + for p in profiles + if not p.hookless and p.hook_scope == "project" + } patterns |= {f"/{rel}" for rel in seeded} _worktree_local_exclude(worktree, sorted(patterns)) @@ -667,15 +733,21 @@ def install_into( bmad_loop_dir = project / ".bmad-loop" bmad_loop_dir.mkdir(parents=True, exist_ok=True) - # 1. hook relay script (shared by all CLIs) - script_target = project / HOOK_SCRIPT_REL - script_source = resources.files("bmad_loop.data").joinpath("bmad_loop_hook.py") - script_target.write_text(script_source.read_text(encoding="utf-8"), encoding="utf-8") - print(f" hook script: {script_target}") + # 1. project-scoped coding CLIs share the copied hook script. Hermes uses + # the package-owned `bmad-loop relay` command in its user-scoped config. + if any(profile.hook_scope == "project" and not profile.hookless for profile in profiles): + script_target = project / HOOK_SCRIPT_REL + script_source = resources.files("bmad_loop.data").joinpath("bmad_loop_hook.py") + script_target.write_text(script_source.read_text(encoding="utf-8"), encoding="utf-8") + print(f" hook script: {script_target}") # 2. per-CLI hook registration for profile in profiles: - if _register_hooks(project, profile) != 0: + if profile.hook_scope == "user": + status = _register_user_scoped_hooks(profile) + else: + status = _register_hooks(project, profile) + if status != 0: return 1 # 3. bundled skills into each CLI's skill tree (deduped: codex+gemini share diff --git a/src/bmad_loop/probe.py b/src/bmad_loop/probe.py index c246df1e..38315973 100644 --- a/src/bmad_loop/probe.py +++ b/src/bmad_loop/probe.py @@ -58,7 +58,7 @@ from . import sanitize from .adapters.multiplexer import MultiplexerError, get_multiplexer from .adapters.profile import CLIProfile -from .install import merge_hooks, relay_registered +from .install import hooks_registered, merge_hooks from .process_host import get_process_host # cmd_probe catches `probe.LeakDetected` around the renderers, mirroring @@ -404,18 +404,7 @@ def infer_token_schema( def _hooks_registered(project: Path, profile: CLIProfile) -> bool: - config_path = project / profile.hooks.config_path - if not config_path.is_file(): - return False - import json - - try: - config = json.loads(config_path.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError): - return False - if not isinstance(config, dict): - return False - return relay_registered(config, profile.hooks.dialect, profile.hooks.events) + return hooks_registered(project, profile) # ----------------------------------------------------------------- SCAN mode @@ -601,6 +590,14 @@ def probe( dialect=profile.hooks.dialect, declared_events=dict(profile.hooks.events), ) + if profile.hook_scope == "user": + scanned = scan(cli=cli, profile=profile, project=project, hints=hints, pseudo=pseudo) + scanned.mode = "probe" + scanned.warnings.append( + "live probing is unavailable for user-scoped Hermes hooks; " + "use `hermes hooks test post_llm_call` to verify the installed relay" + ) + return scanned finding.flags = run_version_help(binary) # The live probe launches through the selected multiplexer backend (see diff --git a/tests/test_cli.py b/tests/test_cli.py index 80d35c4a..1abc7a9b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1365,6 +1365,23 @@ def test_make_adapters_review_synthesizes_from_spec(project, monkeypatch): assert not isinstance(adapters["triage"], GenericDevAdapter) +def test_make_adapters_dispatches_hermes_profile(project, monkeypatch): + from bmad_loop.adapters.hermes import HermesAdapter, HermesDevAdapter + + monkeypatch.setattr(mux_mod, "_usable", lambda mux: True) + install_bmad_config(project) + _write_policy(project.project, '[adapter]\nname = "hermes"\n') + adapters = cli._make_adapters( + project.project, + project.project / ".bmad-loop" / "runs" / "r", + policy_mod.load(project.project / ".bmad-loop" / "policy.toml"), + ) + + assert isinstance(adapters["dev"], HermesDevAdapter) + assert isinstance(adapters["review"], HermesDevAdapter) + assert isinstance(adapters["triage"], HermesAdapter) + + def test_make_adapters_hookless_synthesizing_roles_get_dev_adapter(project, monkeypatch): """Hookless dev/review (bmad-dev-auto roles) dispatch to OpencodeDevAdapter — the _DevSynthesisMixin composed over the HTTP transport — sharing one diff --git a/tests/test_hermes_adapter.py b/tests/test_hermes_adapter.py new file mode 100644 index 00000000..554c19bd --- /dev/null +++ b/tests/test_hermes_adapter.py @@ -0,0 +1,170 @@ +import shlex +import shutil +import subprocess +import sys +import uuid +from pathlib import Path + +import pytest + +from bmad_loop.adapters.base import SessionSpec +from bmad_loop.adapters.hermes import HermesStartupError +from bmad_loop.adapters.hermes import HermesAdapter +from bmad_loop.adapters.profile import get_profile +from bmad_loop.policy import LimitsPolicy, Policy + + +HAVE_TMUX = sys.platform != "win32" and shutil.which("tmux") is not None + + +class StubMux: + def __init__(self, alive: bool = True): + self.alive = alive + self.sessions: set[str] = set() + self.commands: list[str] = [] + self.envs: list[dict[str, str]] = [] + self.sent: list[tuple[str, str]] = [] + + def has_session(self, name): + return name in self.sessions + + def new_session(self, name, cwd, cols, lines): + self.sessions.add(name) + + def set_session_option(self, name, option, value): + pass + + def new_window(self, session, name, cwd, env, command): + self.envs.append(env) + self.commands.append(command) + return "@hermes" + + def pipe_pane(self, window_id, log_file): + pass + + def list_window_ids(self, session): + return ["@hermes"] if self.alive else [] + + def send_text(self, window_id, text): + self.sent.append((window_id, text)) + + +def make_spec(tmp_path: Path) -> SessionSpec: + return SessionSpec( + task_id="todo-1-dev-1", + role="dev", + prompt="/bmad-dev-auto todo-1", + cwd=tmp_path, + env={"BMAD_LOOP_RUN_DIR": str(tmp_path / "run"), "BMAD_LOOP_TASK_ID": "todo-1-dev-1"}, + model="test-model", + ) + + +def make_adapter(tmp_path: Path, mux: StubMux) -> HermesAdapter: + return HermesAdapter( + run_dir=tmp_path / "run", + policy=Policy(limits=LimitsPolicy()), + profile=get_profile("hermes"), + mux=mux, + ) + + +def test_hermes_command_is_interactive_without_initial_prompt(tmp_path): + adapter = make_adapter(tmp_path, StubMux()) + + assert adapter.build_command(make_spec(tmp_path)) == "hermes --cli --accept-hooks --yolo --model test-model" + + +def test_hermes_start_injects_prompt_after_live_startup(tmp_path, monkeypatch): + mux = StubMux(alive=True) + adapter = make_adapter(tmp_path, mux) + monkeypatch.setattr("bmad_loop.adapters.hermes.time.sleep", lambda _: None) + + handle = adapter.start_session(make_spec(tmp_path)) + + assert mux.commands == ["hermes --cli --accept-hooks --yolo --model test-model"] + assert mux.sent == [(handle.native_id, "Use the $bmad-dev-auto skill now: todo-1")] + + +def test_hermes_start_exposes_project_skill_tree_to_tmux_session(tmp_path, monkeypatch): + mux = StubMux(alive=True) + adapter = make_adapter(tmp_path, mux) + monkeypatch.setattr("bmad_loop.adapters.hermes.time.sleep", lambda _: None) + + adapter.start_session(make_spec(tmp_path)) + + assert mux.envs[0]["HERMES_PROJECT_SKILLS"] == str((tmp_path / ".agents" / "skills").resolve()) + + +def test_hermes_start_preserves_explicit_project_skills_env(tmp_path, monkeypatch): + mux = StubMux(alive=True) + adapter = make_adapter(tmp_path, mux) + monkeypatch.setattr("bmad_loop.adapters.hermes.time.sleep", lambda _: None) + spec = make_spec(tmp_path) + explicit = tmp_path / "shared-skills" + spec = SessionSpec( + task_id=spec.task_id, + role=spec.role, + prompt=spec.prompt, + cwd=spec.cwd, + env={**spec.env, "HERMES_PROJECT_SKILLS": str(explicit)}, + model=spec.model, + ) + + adapter.start_session(spec) + + assert mux.envs[0]["HERMES_PROJECT_SKILLS"] == str(explicit) + + +def test_hermes_start_fails_when_pane_dies_during_startup(tmp_path, monkeypatch): + adapter = make_adapter(tmp_path, StubMux(alive=False)) + monkeypatch.setattr("bmad_loop.adapters.hermes.time.sleep", lambda _: None) + + with pytest.raises(HermesStartupError, match="exited during Hermes startup"): + adapter.start_session(make_spec(tmp_path)) + + +@pytest.mark.skipif(not HAVE_TMUX, reason="tmux not available") +def test_hermes_tmux_session_injects_prompt_and_relays_stop(tmp_path, monkeypatch): + fake = tmp_path / "fake-hermes" + fake.write_text( + "#!/bin/bash\n" + "IFS= read -r prompt\n" + "mkdir -p \"$BMAD_LOOP_RUN_DIR/tasks/$BMAD_LOOP_TASK_ID\"\n" + "printf '{\"workflow\": \"fake-hermes\", \"prompt\": \"%s\"}\\n' \"$prompt\" " + "> \"$BMAD_LOOP_RUN_DIR/tasks/$BMAD_LOOP_TASK_ID/result.json\"\n" + "printf '{\"session_id\": \"fake-hermes-session\"}\\n' | " + f"{shlex.quote(sys.executable)} -c " + "'from bmad_loop.cli import main; raise SystemExit(main([\"relay\", \"Stop\"]))'\n" + "sleep 60\n", + encoding="utf-8", + ) + fake.chmod(0o755) + run_dir = tmp_path / ".bmad-loop" / "runs" / f"hermes-{uuid.uuid4().hex[:8]}" + adapter = HermesAdapter( + run_dir=run_dir, + policy=Policy(limits=LimitsPolicy()), + profile=get_profile("hermes"), + binary=str(fake), + extra_args=(), + ) + monkeypatch.setattr(adapter, "STARTUP_GRACE_S", 0.05) + spec = SessionSpec( + task_id="hermes-tmux-e2e", + role="dev", + prompt="/bmad-dev-auto todo-1", + cwd=tmp_path, + env={"BMAD_LOOP_RUN_DIR": str(run_dir), "BMAD_LOOP_TASK_ID": "hermes-tmux-e2e"}, + timeout_s=30, + ) + try: + result = adapter.run(spec) + finally: + subprocess.run(["tmux", "kill-session", "-t", adapter.session_name], capture_output=True) + + assert result.status == "completed" + assert result.session_id == "fake-hermes-session" + assert result.result_json == { + "workflow": "fake-hermes", + "prompt": "Use the $bmad-dev-auto skill now: todo-1", + } diff --git a/tests/test_hermes_hooks.py b/tests/test_hermes_hooks.py new file mode 100644 index 00000000..bb0f0de8 --- /dev/null +++ b/tests/test_hermes_hooks.py @@ -0,0 +1,77 @@ +import io +import json + +from bmad_loop import cli +from bmad_loop.hermes_hooks import ( + HERMES_RELAY_COMMAND, + hermes_config_path, + merge_hermes_stop_hook, + relay_event, + remove_hermes_stop_hook, +) + + +def test_hermes_config_path_honors_hermes_home(tmp_path): + assert hermes_config_path({"HERMES_HOME": str(tmp_path)}) == tmp_path / "config.yaml" + + +def test_merge_hermes_stop_hook_preserves_unrelated_entries_and_is_idempotent(): + original = {"hooks": {"post_llm_call": [{"command": "other-hook", "timeout": 5}]}} + + merged, changed = merge_hermes_stop_hook(original) + + assert changed is True + assert merged["hooks"]["post_llm_call"] == [ + {"command": "other-hook", "timeout": 5}, + {"command": HERMES_RELAY_COMMAND, "timeout": 10}, + ] + repeated, changed = merge_hermes_stop_hook(merged) + assert changed is False + assert repeated == merged + + +def test_remove_hermes_stop_hook_keeps_unrelated_entries(): + config = { + "hooks": { + "post_llm_call": [ + {"command": "other-hook", "timeout": 5}, + {"command": HERMES_RELAY_COMMAND, "timeout": 10}, + ] + } + } + + updated, changed = remove_hermes_stop_hook(config) + + assert changed is True + assert updated == {"hooks": {"post_llm_call": [{"command": "other-hook", "timeout": 5}]}} + + +def test_relay_noops_without_active_bmad_environment(tmp_path): + assert relay_event("Stop", {"session_id": "s-1"}, {}) is False + assert not (tmp_path / "events").exists() + + +def test_relay_writes_stop_event_for_active_bmad_environment(tmp_path): + environ = {"BMAD_LOOP_RUN_DIR": str(tmp_path), "BMAD_LOOP_TASK_ID": "todo-1-dev-1"} + + assert relay_event("Stop", {"session_id": "s-1", "cwd": "/project"}, environ) is True + + event_path = next((tmp_path / "events").glob("*-todo-1-dev-1-Stop.json")) + event = json.loads(event_path.read_text(encoding="utf-8")) + assert event["event"] == "Stop" + assert event["task_id"] == "todo-1-dev-1" + assert event["session_id"] == "s-1" + assert event["transcript_path"] is None + assert event["cwd"] == "/project" + assert isinstance(event["ts"], int) + + +def test_relay_cli_reads_payload_from_stdin(tmp_path, monkeypatch): + monkeypatch.setenv("BMAD_LOOP_RUN_DIR", str(tmp_path)) + monkeypatch.setenv("BMAD_LOOP_TASK_ID", "todo-1-dev-1") + monkeypatch.setattr("sys.stdin", io.StringIO('{"session_id": "s-1", "cwd": "/project"}')) + + assert cli.main(["relay", "Stop"]) == 0 + + event_path = next((tmp_path / "events").glob("*-todo-1-dev-1-Stop.json")) + assert json.loads(event_path.read_text(encoding="utf-8"))["session_id"] == "s-1" diff --git a/tests/test_install.py b/tests/test_install.py index 839050f8..8e423acc 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -12,6 +12,7 @@ LEGACY_MODULE_SKILLS, MODULE_SKILLS, _copy_traversable, + hooks_registered, install_into, merge_hooks, missing_base_skills, @@ -37,6 +38,38 @@ def _registrations(profile, command="python3 /x/.bmad-loop/bmad_loop_hook.py {ev } +def test_install_hermes_registers_one_user_scoped_relay(tmp_path, monkeypatch): + hermes_home = tmp_path / "hermes-home" + hermes_home.mkdir() + config_path = hermes_home / "config.yaml" + config_path.write_text( + "hooks:\n post_llm_call:\n - command: unrelated-hook\n timeout: 5\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert install_into(tmp_path, clis=("hermes",), skills=False) == 0 + assert install_into(tmp_path, clis=("hermes",), skills=False) == 0 + + config = config_path.read_text(encoding="utf-8") + assert "command: unrelated-hook" in config + assert config.count("command: bmad-loop relay Stop") == 1 + assert not (tmp_path / "config.yaml").exists() + assert not (tmp_path / ".bmad-loop" / "bmad_loop_hook.py").exists() + + +def test_hermes_user_scoped_relay_registration_is_detected(tmp_path, monkeypatch): + hermes_home = tmp_path / "hermes-home" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "hooks:\n post_llm_call:\n - command: bmad-loop relay Stop\n timeout: 10\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert hooks_registered(tmp_path, get_profile("hermes")) is True + + def test_merge_hooks_adds_all_events(): profile = get_profile("claude") settings, changed = merge_hooks({}, _registrations(profile), profile.hooks.dialect) @@ -551,6 +584,20 @@ def test_provision_worktree_lays_down_skills_and_hook(tmp_path): assert not (wt / ".bmad-loop").exists() +def test_provision_worktree_hermes_preserves_project_config_yaml(tmp_path): + """Hermes hooks live in HERMES_HOME, never a worktree's application config.""" + wt, repo = tmp_path / "wt", tmp_path / "repo" + project_config = wt / "config.yaml" + project_config.parent.mkdir(parents=True) + project_config.write_text("app:\n name: keep-me\n", encoding="utf-8") + git(wt, "init", "-q") + + provision_worktree(wt, [get_profile("hermes")], repo) + + assert project_config.read_text(encoding="utf-8") == "app:\n name: keep-me\n" + assert "/config.yaml" not in (wt / ".git" / "info" / "exclude").read_text(encoding="utf-8") + + def test_provision_worktree_covers_multiple_profiles(tmp_path): """Dev=claude + review=codex provisions both skill trees (.claude/skills and .agents/skills) and both hook configs.""" diff --git a/tests/test_probe.py b/tests/test_probe.py index 8b6f1255..6d4783b1 100644 --- a/tests/test_probe.py +++ b/tests/test_probe.py @@ -57,6 +57,30 @@ def _write_jsonl(path, rows): ] +def test_live_probe_for_user_scoped_hermes_falls_back_to_scan(tmp_path, monkeypatch): + profile = get_profile("hermes") + scanned = probe.ProfileFinding( + cli="hermes", + mode="scan", + known_profile=True, + binary="hermes", + parser="none", + ) + monkeypatch.setattr(probe, "scan", lambda **_kwargs: scanned) + monkeypatch.setattr( + probe, + "merge_hooks", + lambda *_args: pytest.fail("a user-scoped Hermes config must not enter the JSON probe path"), + ) + + result = probe.probe( + cli="hermes", profile=profile, project=tmp_path, hints=probe.Hints(), timeout_s=1 + ) + + assert result.mode == "probe" + assert any("user-scoped Hermes hooks" in warning for warning in result.warnings) + + # ----------------------------------------------------------- token inference diff --git a/tests/test_profile.py b/tests/test_profile.py index bbccd63f..ea454445 100644 --- a/tests/test_profile.py +++ b/tests/test_profile.py @@ -90,6 +90,29 @@ def test_builtin_profiles_load(): assert profiles[name].hookless is False +def test_builtin_hermes_profile_uses_interactive_user_scoped_adapter(): + profile = load_profiles()["hermes"] + + assert profile.binary == "hermes" + assert profile.adapter == "hermes" + assert profile.hook_scope == "user" + assert profile.skill_tree == ".agents/skills" + assert profile.launch_args == ("--cli", "--accept-hooks", "--yolo") + assert profile.hooks.dialect == "hermes-config-yaml" + assert profile.hooks.events == {"post_llm_call": "Stop"} + + +@pytest.mark.parametrize("field, value", [("adapter", "unknown"), ("hook_scope", "system")]) +def test_profile_rejects_unknown_adapter_and_hook_scope(tmp_path, field, value): + profiles_dir = tmp_path / ".bmad-loop" / "profiles" + profiles_dir.mkdir(parents=True) + profile = MINIMAL_PROFILE.replace("[hooks]", f'{field} = "{value}"\n[hooks]') + (profiles_dir / "invalid.toml").write_text(profile, encoding="utf-8") + + with pytest.raises(ProfileError): + load_profiles(tmp_path) + + def test_usage_grace_and_nudges_default_when_unset(tmp_path): # MINIMAL_PROFILE omits both -> 0.0 / None profiles_dir = tmp_path / ".bmad-loop" / "profiles"