From ccd29d8c5abef4d2cd0fed64502e986d02c27a67 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 21 Sep 2026 14:22:24 -0700 Subject: [PATCH 1/4] Fix Codex hook trust diagnostics (#461) --- CHANGELOG.md | 4 + docs/FEATURES.md | 8 + src/bmad_loop/checks.py | 1 + src/bmad_loop/cli.py | 43 ++++- src/bmad_loop/codex_trust.py | 229 +++++++++++++++++++++++ src/bmad_loop/probe.py | 37 ++++ tests/test_cli.py | 11 +- tests/test_codex_trust.py | 340 +++++++++++++++++++++++++++++++++++ 8 files changed, 667 insertions(+), 6 deletions(-) create mode 100644 src/bmad_loop/codex_trust.py create mode 100644 tests/test_codex_trust.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 00db6b699..c32ce67fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,10 @@ breaking changes may land in a minor release. ### Fixed +- Report stale or unverifiable Codex hook trust in `validate` and `probe-adapter` + before a live probe launches; check both relay events against Codex's read-only + hook discovery for the operation's directory and executable (#461). + - Distinguish confirmed missing tmux-family sessions from failed window listings; raise on unproven liveness failures and warn when metadata uses a sentinel (#525). diff --git a/docs/FEATURES.md b/docs/FEATURES.md index 6b9bf36c9..f9c44dd19 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -702,6 +702,14 @@ persisted artifacts. - Add a CLI without touching Python: drop a TOML profile in `.bmad-loop/profiles/.toml` (binary, prompt template, bypass flags, hook dialect, native→canonical event map). A CLI that needs its own adapter _class_ still needs Python — but not a core edit: the profile's `adapter` field names a kind resolved against the registry, which a co-installed package extends. - `bmad-loop probe-adapter` collects + sanitizes the data needed to finalize/add a profile (hook payload shape, transcript location/format, token schema): a zero-launch scan by default, opt-in `--probe` for live capture. See the [adapter authoring guide](adapter-authoring-guide.md). +For Codex, `validate` and `probe-adapter` ask Codex's read-only `hooks/list` API +whether the configured SessionStart and Stop relays are enabled and trusted. +Stale, missing, or unverifiable hook trust is a failing result. A worktree run +uses a different directory, so `validate` cannot certify its future trust from +the main checkout. A live probe checks its temporary hook directory before +launch; a fresh directory without a Codex trust grant stops with a hook-trust +diagnostic. + ### Budgeting & cost tracking - Mid-session per-session token budget (`max_tokens_per_session`, default 4M weighted): both adapter wait loops sample cumulative usage on the ~30s heartbeat and trip once on crossing, per `session_budget_mode` — `warn` (default) raises an ATTENTION + lifecycle breadcrumb only; `enforce` also sends a wrap-up nudge, grants `session_budget_grace_s` (default 240s) to finish, then terminates the session `over_budget` (ordinary retry→defer routing; an artifact flushed at kill time is still honored). Sampling is live-verified on `claude` and best-effort on other transcript-reading profiles (two independent unknowns there: whether the CLI delivers the transcript path early — until a hook event carries it the guard is inert — and whether it flushes usage mid-turn); the nudge into a busy pane is best-effort everywhere (the termination is the guarantee), and adapters with no mid-session usage signal (`usage_parser = "none"`, Copilot's shutdown-only flush) leave the guard inert. diff --git a/src/bmad_loop/checks.py b/src/bmad_loop/checks.py index e44932bd3..d5029e117 100644 --- a/src/bmad_loop/checks.py +++ b/src/bmad_loop/checks.py @@ -66,6 +66,7 @@ "git.version", "hooks.config-parse", "hooks.registered", + "hooks.trust", "hooks.relay-present", "hooks.relay-stale", "mux.backend", diff --git a/src/bmad_loop/cli.py b/src/bmad_loop/cli.py index 7cb3b94b6..14a2afae8 100644 --- a/src/bmad_loop/cli.py +++ b/src/bmad_loop/cli.py @@ -741,6 +741,37 @@ def cmd_validate(args: argparse.Namespace) -> int: {"profile": profile.name, "config_path": str(hook_config)}, ) + if profile.hooks.dialect == "codex-hooks-json": + from .codex_trust import project_hook_trust + + if not profile.packaged: + trust_message = ( + "hook trust unverifiable: project-owned Codex profile may name an " + "untrusted executable; validation will not launch it" + ) + elif pol is not None and pol.scm.isolation == "worktree": + trust_message = ( + "hook trust unverifiable for future worktree sessions: each isolated " + "directory needs its own Codex trust grant" + ) + elif not hooks_ok: + trust_message = "hook trust cannot pass: Codex relay hooks are not registered" + else: + trust = project_hook_trust(project, profile) + trust_message = None if trust.status == "trusted" else trust.reason + if trust_message is None: + report.ok( + "hooks.trust", + f"Codex hook trust current for {profile.name} in {project}", + {"profile": profile.name, "project": str(project), "binary": profile.binary}, + ) + else: + report.fail( + "hooks.trust", + f"{profile.name}: {trust_message}", + {"profile": profile.name, "project": str(project), "binary": profile.binary}, + ) + # #461: `hooks.registered` above is a substring match on the config JSON — it # never touches the artifact the registered command points AT. A branch switch # (or a deleted .bmad-loop/) leaves the registration green while every hook @@ -3106,8 +3137,7 @@ def _resume_paused_run(project: Path, run_dir: Path) -> int: # instead of reloading the predecessor's old paused state and double-driving. if runs.engine_liveness(run_dir) == "alive": print( - f"run {run_dir.name} is still live — resuming would double-drive it; " - "stop it first", + f"run {run_dir.name} is still live — resuming would double-drive it; stop it first", file=sys.stderr, ) return 1 @@ -5104,6 +5134,8 @@ def cmd_probe(args: argparse.Namespace) -> int: # Every `ok:` trailer is human-facing chatter, so in JSON mode it goes to # stderr — stdout is the document alone, or empty when --out took it. trailers = sys.stderr if args.json else sys.stdout + trust_ok = finding.hook_trust is None or finding.hook_trust == "trusted" + trailer_prefix = "ok" if trust_ok else "FAIL" if args.out: out_path = Path(args.out) if args.json: @@ -5111,7 +5143,7 @@ def cmd_probe(args: argparse.Namespace) -> int: else: out_path.write_text(report, encoding="utf-8") print( - f" ok: {noun} written to {out_path} ({len(finding.warnings)} warning(s))", + f" {trailer_prefix}: {noun} written to {out_path} ({len(finding.warnings)} warning(s))", file=trailers, ) else: @@ -5120,10 +5152,11 @@ def cmd_probe(args: argparse.Namespace) -> int: else: print(report) print( - f" ok: {finding.mode} {noun} for {args.cli} ({len(finding.warnings)} warning(s))", + f" {trailer_prefix}: {finding.mode} {noun} for {args.cli} " + f"({len(finding.warnings)} warning(s))", file=trailers, ) - return 0 + return 0 if trust_ok else 1 def cmd_diagnose(args: argparse.Namespace) -> int: diff --git a/src/bmad_loop/codex_trust.py b/src/bmad_loop/codex_trust.py new file mode 100644 index 000000000..fbbf71a93 --- /dev/null +++ b/src/bmad_loop/codex_trust.py @@ -0,0 +1,229 @@ +"""Read Codex's own hook trust verdict without starting a model turn. + +The private trust hash belongs to Codex. This module asks its read-only +``hooks/list`` app-server method and joins that answer to the exact commands in +the hook config at the directory an operation will use. +""" + +from __future__ import annotations + +import json +import os +import queue +import subprocess +import threading +import time +from dataclasses import dataclass +from pathlib import Path + +from .adapters.profile import CLIProfile + +_EVENTS = {"SessionStart": "sessionStart", "Stop": "stop"} +_RELAY_MARKER = "bmad_loop_hook.py" +_PROBE_MARKER = "bmad_loop_probe_hook.py" +_TIMEOUT_S = 5.0 + + +@dataclass(frozen=True) +class TrustResult: + status: str # trusted | untrusted | unverifiable + reason: str + + +def _commands(config: object, profile: CLIProfile, marker: str) -> dict[str, list[str]] | None: + if not isinstance(config, dict) or not isinstance(config.get("hooks"), dict): + raise ValueError("malformed Codex hook config") + events = profile.hooks.events + if not all(events.get(event) == event for event in _EVENTS): + return None + found: dict[str, list[str]] = {} + for canonical in _EVENTS: + handlers = config["hooks"].get(canonical) + if handlers is None: + return None + if not isinstance(handlers, list): + raise ValueError("malformed Codex hook handlers") + commands: list[str] = [] + for group in handlers: + if not isinstance(group, dict) or not isinstance(group.get("hooks"), list): + raise ValueError("malformed Codex hook group") + for hook in group["hooks"]: + if not isinstance(hook, dict): + raise ValueError("malformed Codex hook entry") + command = hook.get("command") + if isinstance(command, str) and marker in command: + commands.append(command) + if not commands: + return None + found[canonical] = commands + return found + + +def _hooks_list(binary: str, cwd: Path, env: dict[str, str]) -> object: + """Execute initialize → initialized → hooks/list with one absolute deadline.""" + child = subprocess.Popen( + [binary, "app-server", "--stdio"], + cwd=cwd, + env={**os.environ, **env}, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + ) + stdin, stdout = child.stdin, child.stdout + assert stdin is not None and stdout is not None + lines: queue.Queue[str] = queue.Queue() + + def read_lines() -> None: + for line in stdout: + lines.put(line) + lines.put("") + + threading.Thread(target=read_lines, daemon=True).start() + deadline = time.monotonic() + _TIMEOUT_S + + def send(message: dict) -> None: + stdin.write(json.dumps(message) + "\n") + stdin.flush() + + def response(identifier: int) -> object: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Codex hooks/list timed out") + try: + line = lines.get(timeout=remaining) + except queue.Empty as exc: + raise TimeoutError("Codex hooks/list timed out") from exc + if not line: + raise ValueError("Codex app server closed before hooks/list") + try: + message = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError("Codex app server returned invalid JSON") from exc + if not isinstance(message, dict) or message.get("id") != identifier: + continue # notifications and unrelated messages + if "error" in message: + raise ValueError("Codex app server refused hooks/list") + if "result" not in message: + raise ValueError("Codex app server omitted a result") + return message["result"] + + try: + send( + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "clientInfo": {"name": "bmad-loop", "version": "0"}, + "capabilities": {"experimentalApi": True}, + }, + } + ) + response(1) + send({"jsonrpc": "2.0", "method": "initialized", "params": {}}) + send( + { + "jsonrpc": "2.0", + "id": 2, + "method": "hooks/list", + "params": {"cwds": [str(cwd.resolve())]}, + } + ) + return response(2) + finally: + child.kill() + child.wait(timeout=2) + + +def project_hook_trust( + project: Path, + profile: CLIProfile, + *, + binary: str | None = None, + marker: str = _RELAY_MARKER, +) -> TrustResult: + """Fail closed unless Codex reports every required configured hook trusted.""" + if profile.hooks.dialect != "codex-hooks-json": + return TrustResult("unverifiable", "hook trust applies only to Codex hooks") + # The default bypass switch changes approvals, not hook configuration. Other + # launch arguments can select a different config and cannot be mirrored here. + if profile.launch_args or any( + arg != "--dangerously-bypass-approvals-and-sandbox" for arg in profile.bypass_args + ): + return TrustResult("unverifiable", "hook trust cannot verify profile launch arguments") + config_path = (project / profile.hooks.config_path).resolve() + try: + config = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, ValueError): + return TrustResult("unverifiable", "hook trust config is unreadable") + try: + commands = _commands(config, profile, marker) + except ValueError: + return TrustResult("unverifiable", "hook trust config has malformed fields") + if commands is None: + return TrustResult( + "untrusted", "hook trust: a required SessionStart or Stop hook is not registered" + ) + try: + result = _hooks_list(binary or profile.binary, project, profile.env) + except (OSError, ValueError, TimeoutError, subprocess.SubprocessError): + return TrustResult("unverifiable", "hook trust could not be queried from Codex") + if not isinstance(result, dict) or not isinstance(result.get("data"), list): + return TrustResult("unverifiable", "hook trust response has an unfamiliar shape") + entries = result["data"] + if len(entries) != 1 or not isinstance(entries[0], dict): + return TrustResult("unverifiable", "hook trust response has an unfamiliar directory") + entry = entries[0] + if entry.get("cwd") != str(project.resolve()): + return TrustResult("unverifiable", "hook trust response names another directory") + if ( + not isinstance(entry.get("errors"), list) + or not isinstance(entry.get("hooks"), list) + or not isinstance(entry.get("warnings"), list) + ): + return TrustResult("unverifiable", "hook trust response has malformed fields") + if entry["errors"] or entry["warnings"]: + return TrustResult("unverifiable", "hook trust discovery reported errors or warnings") + hooks = entry["hooks"] + for hook in hooks: + if ( + not isinstance(hook, dict) + or not all( + isinstance(hook.get(key), str) for key in ("sourcePath", "eventName", "trustStatus") + ) + or not isinstance(hook.get("enabled"), bool) + ): + return TrustResult("unverifiable", "hook trust response contains a malformed hook") + for canonical, expected in commands.items(): + matches: list[dict] = [] + for hook in hooks: + if ( + hook.get("sourcePath") != str(config_path) + or hook.get("eventName") != _EVENTS[canonical] + ): + continue + if hook.get("handlerType") != "command" or not isinstance(hook.get("command"), str): + return TrustResult( + "unverifiable", "hook trust response has unfamiliar handler fields" + ) + matches.append(hook) + for command in expected: + candidates = [h for h in matches if h["command"] == command] + if len(candidates) != 1: + return TrustResult("untrusted", f"hook trust: Codex omitted the {canonical} relay") + hook = candidates[0] + status = hook.get("trustStatus") + if not isinstance(status, str) or not isinstance(hook.get("enabled"), bool): + return TrustResult("unverifiable", "hook trust response has malformed trust fields") + if status not in {"trusted", "managed", "modified", "untrusted"}: + return TrustResult("unverifiable", "hook trust response has unfamiliar status") + if status not in {"trusted", "managed"} or not hook["enabled"]: + return TrustResult( + "untrusted", f"hook trust is stale for {canonical}; accept hooks in Codex" + ) + return TrustResult("trusted", "Codex hook trust current for SessionStart and Stop") diff --git a/src/bmad_loop/probe.py b/src/bmad_loop/probe.py index 23ae8c854..4181e42d3 100644 --- a/src/bmad_loop/probe.py +++ b/src/bmad_loop/probe.py @@ -162,6 +162,7 @@ class ProfileFinding: flags: FlagFinding | None = None declared_events: dict = field(default_factory=dict) # native -> canonical registered: bool | None = None # scan: hooks present in the CLI's config? + hook_trust: str | None = None # trusted | untrusted | unverifiable (Codex only) captured_events: list[EventCapture] = field(default_factory=list) # probe transcript: TranscriptFinding | None = None tokens: TokenSchema | None = None @@ -473,6 +474,32 @@ def _hooks_registered(project: Path, profile: CLIProfile) -> bool: return relay_registered(config, profile.hooks.dialect, profile.hooks.events) +def _check_hook_trust( + finding: ProfileFinding, project: Path, profile: CLIProfile, binary: str, *, live: bool = False +) -> None: + if profile.hooks.dialect != "codex-hooks-json": + return + from .codex_trust import project_hook_trust + + marker = PROBE_HOOK_NAME if live else "bmad_loop_hook.py" + trust = project_hook_trust(project, profile, binary=binary, marker=marker) + finding.hook_trust = trust.status + if trust.status != "trusted": + scope = "temporary probe workspace" if live else "project checkout" + finding.warnings.append( + f"Codex hook trust for {scope} using {finding.binary}: {trust.reason}" + ) + if live: + finding.next_steps.append( + "Codex has no trust grant for this fresh temporary workspace; " + "live capture cannot proceed until that workspace is trusted" + ) + else: + finding.next_steps.append( + "Open Codex in the project checkout and accept its hook trust prompt" + ) + + # ----------------------------------------------------------------- SCAN mode @@ -513,6 +540,7 @@ def scan( if profile is not None: finding.registered = _hooks_registered(project, profile) + _check_hook_trust(finding, project, profile, binary) if not finding.registered: finding.next_steps.append( f"hooks not registered in {profile.hooks.config_path}; " @@ -708,6 +736,12 @@ def probe( config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8") + # The source checkout's trust says nothing about this new directory. + # Query after writing its config and before launching a model turn. + _check_hook_trust(finding, tmpdir, profile, binary, live=True) + if finding.hook_trust is not None and finding.hook_trust != "trusted": + return finding + # 2. launch one trivial content-free turn in a fresh tmux window argv = _probe_argv(profile, binary, hints) env = { @@ -843,6 +877,8 @@ def render_markdown( out.append(_fmt_kv("usage_parser", f.parser)) if f.registered is not None: out.append(_fmt_kv("hooks registered", "yes" if f.registered else "no")) + if f.hook_trust is not None: + out.append(_fmt_kv("Codex hook trust", f.hook_trust)) out.append(_fmt_kv("warnings", str(len(f.warnings)))) out.append("") @@ -993,6 +1029,7 @@ def transcript_dict(t: TranscriptFinding | None): "dialect": f.dialect, "usage_parser": f.parser, "hooks_registered": f.registered, + "hook_trust": f.hook_trust, "declared_events": f.declared_events, "version": f.flags.version if f.flags else None, "help": f.flags.help if f.flags else None, diff --git a/tests/test_cli.py b/tests/test_cli.py index f040c624f..50bd093c4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -9903,7 +9903,7 @@ def test_validate_stories_folder_known_selector_ok(project): def _make_validate_pass(project, monkeypatch, capsys, *, policy=CLAUDE_ONLY_POLICY, skills=None): - """Set a project up so every validate gate passes, and pin the three gates whose + """Set a project up so every validate gate passes, and pin the gates whose outcome is a property of the *host* rather than of the project: whether the CLI binary is on PATH, whether it actually runs, and whether a multiplexer is installed. Without those pins the rc-0 leg would pass or fail by machine, which @@ -9933,6 +9933,15 @@ def _make_validate_pass(project, monkeypatch, capsys, *, policy=CLAUDE_ONLY_POLI git(project.project, "commit", "-q", "-m", "validate fixture") monkeypatch.setattr(cli.shutil, "which", lambda tool: f"/usr/bin/{tool}") monkeypatch.setattr(probe_mod, "binary_runs", lambda *_a, **_kw: 0) + # A sandbox created during the test has no Codex trust grant, even when its + # registration is valid. Trust classification itself is tested separately. + from bmad_loop import codex_trust + + monkeypatch.setattr( + codex_trust, + "project_hook_trust", + lambda *_a, **_kw: codex_trust.TrustResult("trusted", "fixture trust grant"), + ) monkeypatch.setattr( cli, "_platform_preflight", diff --git a/tests/test_codex_trust.py b/tests/test_codex_trust.py new file mode 100644 index 000000000..7031d8240 --- /dev/null +++ b/tests/test_codex_trust.py @@ -0,0 +1,340 @@ +"""Codex trust is checked at the same executable and directory as a session.""" + +from __future__ import annotations + +import json +import sys +from dataclasses import replace +from pathlib import Path + +import pytest +from conftest import install_bmad_config + +from bmad_loop import cli, codex_trust, probe +from bmad_loop.adapters.profile import get_profile +from bmad_loop.install import merge_hooks + + +def _config(root: Path, commands: dict[str, str] | None = None) -> dict: + profile = get_profile("codex") + if commands is None: + commands = { + event: f"python3 {root}/.bmad-loop/bmad_loop_hook.py {event}" + for event in ("SessionStart", "Stop") + } + data, _ = merge_hooks({}, commands, profile.hooks.dialect) + path = root / profile.hooks.config_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data), encoding="utf-8") + return data + + +def _rpc(root: Path, data: dict, status: str = "trusted") -> dict: + hooks = [ + { + "sourcePath": str((root / ".codex/hooks.json").resolve()), + "eventName": event[0].lower() + event[1:], + "handlerType": "command", + "command": data["hooks"][event][0]["hooks"][0]["command"], + "enabled": True, + "trustStatus": status, + } + for event in ("SessionStart", "Stop") + ] + return {"data": [{"cwd": str(root.resolve()), "errors": [], "warnings": [], "hooks": hooks}]} + + +def test_scripted_app_server_executes_request_sequence_and_reads_environment(tmp_path): + """An actual zero-token child parses initialize and hooks/list, not a mocked RPC.""" + data = _config(tmp_path) + reply = _rpc(tmp_path, data) + script = tmp_path / "codex-stub" + log = tmp_path / "requests.json" + script.write_text( + f"#!{sys.executable}\n" + "import json, os, sys\n" + "requests = []\n" + "for line in sys.stdin:\n" + " msg = json.loads(line); requests.append(msg)\n" + " if msg.get('method') == 'initialize':\n" + " print(json.dumps({'id': 1, 'result': {}}), flush=True)\n" + " if msg.get('method') == 'hooks/list':\n" + " assert os.environ['CODEX_HOME'] == 'test-home'\n" + f" open({str(log)!r}, 'w').write(json.dumps(requests))\n" + f" print(json.dumps({{'id': 2, 'result': {reply!r}}}), flush=True)\n", + encoding="utf-8", + ) + script.chmod(0o755) + profile = replace(get_profile("codex"), binary=str(script), env={"CODEX_HOME": "test-home"}) + result = codex_trust.project_hook_trust(tmp_path, profile) + assert result.status == "trusted", result.reason + requests = json.loads(log.read_text(encoding="utf-8")) + assert [item["method"] for item in requests] == ["initialize", "initialized", "hooks/list"] + assert requests[-1]["params"]["cwds"] == [str(tmp_path.resolve())] + + +@pytest.mark.parametrize( + ("mutation", "expected"), + [ + ("start-modified", "untrusted"), + ("start-omitted", "untrusted"), + ("modified", "untrusted"), + ("disabled", "untrusted"), + ("omitted", "untrusted"), + ("wrong-command", "untrusted"), + ("malformed-status", "unverifiable"), + ("wrong-source", "untrusted"), + ], +) +def test_trust_refuses_stale_or_unmatched_relay(tmp_path, monkeypatch, mutation, expected): + data = _config(tmp_path) + result = _rpc(tmp_path, data) + hooks = result["data"][0]["hooks"] + if mutation == "start-modified": + hooks[0]["trustStatus"] = "modified" + elif mutation == "start-omitted": + hooks.pop(0) + elif mutation == "modified": + hooks[1]["trustStatus"] = "modified" + elif mutation == "disabled": + hooks[1]["enabled"] = False + elif mutation == "omitted": + hooks.pop() + elif mutation == "wrong-command": + hooks[1]["command"] = "echo unrelated" + elif mutation == "malformed-status": + hooks[1]["trustStatus"] = [] + elif mutation == "wrong-source": + hooks[1]["sourcePath"] = "/tmp/other/hooks.json" + monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: result) + assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == expected + + +def test_missing_profile_stop_and_unsupported_launch_args_fail_closed(tmp_path, monkeypatch): + data = _config(tmp_path) + monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: _rpc(tmp_path, data)) + profile = get_profile("codex") + assert ( + codex_trust.project_hook_trust(tmp_path, replace(profile, launch_args=("-c", "x=1"))).status + == "unverifiable" + ) + assert ( + codex_trust.project_hook_trust( + tmp_path, replace(profile, env={"CODEX_HOME": "another"}) + ).status + == "trusted" + ) + config_path = tmp_path / profile.hooks.config_path + config_path.write_text(json.dumps({"hooks": {"SessionStart": data["hooks"]["SessionStart"]}})) + assert codex_trust.project_hook_trust(tmp_path, profile).status == "untrusted" + config_path.write_text(json.dumps(data)) + assert ( + codex_trust.project_hook_trust( + tmp_path, + replace(profile, hooks=replace(profile.hooks, events={"SessionStart": "SessionStart"})), + ).status + == "untrusted" + ) + + +def test_validate_names_untrusted_hook_and_refuses_worktree_inference(project, monkeypatch, capsys): + from bmad_loop.install import install_into + + install_bmad_config(project) + install_into(project.project, clis=("codex",)) + capsys.readouterr() + policy = project.project / ".bmad-loop/policy.toml" + policy.write_text('[adapter]\nname = "codex"\n', encoding="utf-8") + monkeypatch.setattr( + codex_trust, + "project_hook_trust", + lambda *_args, **_kwargs: codex_trust.TrustResult("untrusted", "hook trust stale"), + ) + cli.main(["validate", "--project", str(project.project), "--json"]) + out, err = capsys.readouterr() + assert out, err + doc = json.loads(out) + findings = [f for f in doc["findings"] if f["check"] == "hooks.trust"] + assert len(findings) == 1 and findings[0]["severity"] == "problem" + assert "hook trust stale" in findings[0]["message"] + + policy.write_text( + '[adapter]\nname = "codex"\n[scm]\nisolation = "worktree"\n', encoding="utf-8" + ) + cli.main(["validate", "--project", str(project.project), "--json"]) + doc = json.loads(capsys.readouterr().out) + findings = [f for f in doc["findings"] if f["check"] == "hooks.trust"] + assert len(findings) == 1 and "worktree" in findings[0]["message"] + + +def test_validate_does_not_run_project_owned_codex_profile(project, tmp_path, capsys): + from bmad_loop.install import install_into + + install_bmad_config(project) + install_into(project.project, clis=("codex",)) + capsys.readouterr() + policy = project.project / ".bmad-loop/policy.toml" + policy.write_text('[adapter]\nname = "codex"\n', encoding="utf-8") + sentinel = tmp_path / "executed" + binary = project.project / "codex-stub" + binary.write_text( + f"#!{sys.executable}\nfrom pathlib import Path\nPath({str(sentinel)!r}).write_text('yes')\n", + encoding="utf-8", + ) + binary.chmod(0o755) + overlay = project.project / ".bmad-loop/profiles/codex.toml" + overlay.parent.mkdir(parents=True, exist_ok=True) + overlay.write_text( + f'name = "codex"\nbinary = "{binary}"\n' + '[hooks]\ndialect = "codex-hooks-json"\nconfig_path = ".codex/hooks.json"\n' + 'events = { SessionStart = "SessionStart", Stop = "Stop" }\n', + encoding="utf-8", + ) + cli.main(["validate", "--project", str(project.project), "--json"]) + doc = json.loads(capsys.readouterr().out) + assert not sentinel.exists() + finding = next(f for f in doc["findings"] if f["check"] == "hooks.trust") + assert finding["severity"] == "problem" and "project-owned" in finding["message"] + + +def test_scan_and_live_probe_refuse_trust_at_their_own_directories(tmp_path, monkeypatch): + profile = get_profile("codex") + _config(tmp_path) + calls = [] + + def trust(path, _profile, *, binary=None, marker=None): + calls.append((path, binary, marker)) + return codex_trust.TrustResult("untrusted", "hook trust stale") + + monkeypatch.setattr(codex_trust, "project_hook_trust", trust) + monkeypatch.setattr(probe, "run_version_help", lambda binary: probe.FlagFinding(binary, True)) + scanned = probe.scan( + cli="codex", profile=profile, project=tmp_path, hints=probe.Hints(binary="chosen") + ) + assert scanned.hook_trust == "untrusted" and calls[-1][:2] == (tmp_path, "chosen") + + class Mux: + def available(self): + return True + + class Launcher: + def __init__(self, **_kwargs): + pass + + def start(self, *_args): + pytest.fail("untrusted temporary hook config must stop before launch") + + def kill(self): + pass + + monkeypatch.setattr(probe, "get_multiplexer", Mux) + monkeypatch.setattr(probe, "_ProbeLauncher", Launcher) + monkeypatch.setattr(probe.shutil, "which", lambda _binary: "/bin/true") + live = probe.probe( + cli="codex", profile=profile, project=tmp_path, hints=probe.Hints(binary="chosen") + ) + assert live.hook_trust == "untrusted" + assert calls[-1][0] != tmp_path and calls[-1][1:] == ("chosen", probe.PROBE_HOOK_NAME) + assert "temporary probe workspace" in live.warnings[0] + + +def test_trusted_live_probe_checks_temp_config_then_starts_zero_token_launcher( + tmp_path, monkeypatch +): + profile = get_profile("codex") + events = [] + + def trust(path, _profile, *, binary=None, marker=None): + assert (path / profile.hooks.config_path).is_file() + events.append(("trust", path, binary, marker)) + return codex_trust.TrustResult("trusted", "hook trust current") + + class Mux: + def available(self): + return True + + class Launcher: + def __init__(self, **_kwargs): + pass + + def start(self, argv, _env, cwd, _log_file): + events.append(("start", cwd, argv[0])) + return "fake-window" + + def kill(self): + events.append(("kill",)) + + class Watcher: + def __init__(self, _capture_dir): + pass + + def wait_for(self, *_args, **_kwargs): + return object() # scripted Stop; no model turn + + monkeypatch.setattr(codex_trust, "project_hook_trust", trust) + monkeypatch.setattr(probe, "get_multiplexer", Mux) + monkeypatch.setattr(probe, "_ProbeLauncher", Launcher) + monkeypatch.setattr(probe, "SignalWatcher", Watcher) + monkeypatch.setattr(probe.shutil, "which", lambda _binary: "/bin/true") + monkeypatch.setattr(probe, "run_version_help", lambda binary: probe.FlagFinding(binary, True)) + monkeypatch.setattr(probe, "discover_transcript", lambda *_args, **_kwargs: None) + monkeypatch.setattr(probe.time, "sleep", lambda _seconds: None) + + finding = probe.probe( + cli="codex", profile=profile, project=tmp_path, hints=probe.Hints(binary="chosen") + ) + assert finding.hook_trust == "trusted" + assert events[0][0] == "trust" and events[0][1] != tmp_path + assert events[0][2:] == ("chosen", probe.PROBE_HOOK_NAME) + assert events[1] == ("start", events[0][1], "chosen") + assert events[-1] == ("kill",) + + +def test_probe_json_exits_nonzero_and_names_hook_trust(tmp_path, monkeypatch, capsys): + _config(tmp_path) + monkeypatch.setattr( + codex_trust, + "project_hook_trust", + lambda *_args, **_kwargs: codex_trust.TrustResult("untrusted", "hook trust stale"), + ) + monkeypatch.setattr(probe, "run_version_help", lambda binary: probe.FlagFinding(binary, True)) + monkeypatch.setattr(probe, "discover_transcript", lambda *_args, **_kwargs: None) + rc = cli.main( + ["probe-adapter", "codex", "--project", str(tmp_path), "--binary", "chosen", "--json"] + ) + out, err = capsys.readouterr() + assert rc == 1 and "FAIL" in err + doc = json.loads(out) + assert doc["hook_trust"] == "untrusted" + assert "hook trust stale" in doc["warnings"][0] + + +def test_probe_scan_with_unregistered_codex_hooks_is_non_green(tmp_path, monkeypatch, capsys): + monkeypatch.setattr(probe, "run_version_help", lambda binary: probe.FlagFinding(binary, True)) + monkeypatch.setattr(probe, "discover_transcript", lambda *_args, **_kwargs: None) + rc = cli.main(["probe-adapter", "codex", "--project", str(tmp_path), "--json"]) + doc = json.loads(capsys.readouterr().out) + assert rc == 1 + assert doc["hooks_registered"] is False + assert doc["hook_trust"] != "trusted" + assert any("hook trust" in warning for warning in doc["warnings"]) + + +def test_continuous_unrelated_messages_cannot_extend_rpc_deadline(tmp_path, monkeypatch): + script = tmp_path / "chatty-codex" + script.write_text( + f"#!{sys.executable}\n" + "import json, sys\n" + "for line in sys.stdin:\n" + " message = json.loads(line)\n" + " if message.get('method') == 'initialize':\n" + " print(json.dumps({'id': 1, 'result': {}}), flush=True)\n" + " if message.get('method') == 'hooks/list':\n" + " while True:\n" + " print(json.dumps({'method': 'unrelated'}), flush=True)\n", + encoding="utf-8", + ) + script.chmod(0o755) + monkeypatch.setattr(codex_trust, "_TIMEOUT_S", 0.1) + with pytest.raises(TimeoutError): + codex_trust._hooks_list(str(script), tmp_path, {}) From 1d1008930e7827cda7992b9017e6c113350170f2 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 21 Sep 2026 15:00:24 -0700 Subject: [PATCH 2/4] Address Codex trust review findings --- docs/FEATURES.md | 2 + src/bmad_loop/cli.py | 14 +++++- src/bmad_loop/codex_trust.py | 32 ++++++++++--- tests/test_codex_trust.py | 87 +++++++++++++++++++++++++++--------- 4 files changed, 109 insertions(+), 26 deletions(-) diff --git a/docs/FEATURES.md b/docs/FEATURES.md index f9c44dd19..199b42a11 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -709,6 +709,8 @@ uses a different directory, so `validate` cannot certify its future trust from the main checkout. A live probe checks its temporary hook directory before launch; a fresh directory without a Codex trust grant stops with a hook-trust diagnostic. +Profile or stage arguments that can change Codex hook discovery make the trust +verdict unverifiable rather than certifying a different launch configuration. ### Budgeting & cost tracking diff --git a/src/bmad_loop/cli.py b/src/bmad_loop/cli.py index 14a2afae8..25cfcc06b 100644 --- a/src/bmad_loop/cli.py +++ b/src/bmad_loop/cli.py @@ -742,7 +742,14 @@ def cmd_validate(args: argparse.Namespace) -> int: ) if profile.hooks.dialect == "codex-hooks-json": - from .codex_trust import project_hook_trust + from .codex_trust import hook_discovery_args_safe, project_hook_trust + + unsafe_roles = [] + if pol is not None: + for role in ROLES: + cfg = pol.adapter.resolved(role) + if cfg.name == profile.name and not hook_discovery_args_safe(cfg.extra_args): + unsafe_roles.append(role) if not profile.packaged: trust_message = ( @@ -756,6 +763,11 @@ def cmd_validate(args: argparse.Namespace) -> int: ) elif not hooks_ok: trust_message = "hook trust cannot pass: Codex relay hooks are not registered" + elif unsafe_roles: + trust_message = ( + "hook trust unverifiable: adapter.extra_args may change Codex hook " + f"discovery for {', '.join(unsafe_roles)}" + ) else: trust = project_hook_trust(project, profile) trust_message = None if trust.status == "trusted" else trust.reason diff --git a/src/bmad_loop/codex_trust.py b/src/bmad_loop/codex_trust.py index fbbf71a93..c93bbc5b5 100644 --- a/src/bmad_loop/codex_trust.py +++ b/src/bmad_loop/codex_trust.py @@ -10,6 +10,7 @@ import json import os import queue +import shutil import subprocess import threading import time @@ -17,11 +18,13 @@ from pathlib import Path from .adapters.profile import CLIProfile +from .process_host import ProcessHostError, get_process_host _EVENTS = {"SessionStart": "sessionStart", "Stop": "stop"} _RELAY_MARKER = "bmad_loop_hook.py" _PROBE_MARKER = "bmad_loop_probe_hook.py" _TIMEOUT_S = 5.0 +_SAFE_BYPASS_ARG = "--dangerously-bypass-approvals-and-sandbox" @dataclass(frozen=True) @@ -30,6 +33,11 @@ class TrustResult: reason: str +def hook_discovery_args_safe(args: tuple[str, ...] | None) -> bool: + """Whether extra launch args can leave hook discovery unchanged.""" + return args is None or all(arg == _SAFE_BYPASS_ARG for arg in args) + + def _commands(config: object, profile: CLIProfile, marker: str) -> dict[str, list[str]] | None: if not isinstance(config, dict) or not isinstance(config.get("hooks"), dict): raise ValueError("malformed Codex hook config") @@ -136,7 +144,16 @@ def response(identifier: int) -> object: ) return response(2) finally: - child.kill() + if child.poll() is None: + if os.name == "nt": + # An npm .cmd shim is a cmd.exe parent. Kill its tree before + # the wrapper exits and leaves the app server running. + try: + get_process_host().force_kill(child.pid) + except (OSError, ProcessHostError): + child.kill() + else: + child.kill() child.wait(timeout=2) @@ -152,9 +169,7 @@ def project_hook_trust( return TrustResult("unverifiable", "hook trust applies only to Codex hooks") # The default bypass switch changes approvals, not hook configuration. Other # launch arguments can select a different config and cannot be mirrored here. - if profile.launch_args or any( - arg != "--dangerously-bypass-approvals-and-sandbox" for arg in profile.bypass_args - ): + if profile.launch_args or not hook_discovery_args_safe(profile.bypass_args): return TrustResult("unverifiable", "hook trust cannot verify profile launch arguments") config_path = (project / profile.hooks.config_path).resolve() try: @@ -169,8 +184,15 @@ def project_hook_trust( return TrustResult( "untrusted", "hook trust: a required SessionStart or Stop hook is not registered" ) + # Windows npm installs expose a codex.cmd shim through PATHEXT. Popen with + # a bare name need not find it; which() returns the executable run would use. + resolved_binary = shutil.which( + binary or profile.binary, path={**os.environ, **profile.env}.get("PATH") + ) + if resolved_binary is None: + return TrustResult("unverifiable", "hook trust Codex binary is unavailable") try: - result = _hooks_list(binary or profile.binary, project, profile.env) + result = _hooks_list(resolved_binary, project, profile.env) except (OSError, ValueError, TimeoutError, subprocess.SubprocessError): return TrustResult("unverifiable", "hook trust could not be queried from Codex") if not isinstance(result, dict) or not isinstance(result.get("data"), list): diff --git a/tests/test_codex_trust.py b/tests/test_codex_trust.py index 7031d8240..a678fd611 100644 --- a/tests/test_codex_trust.py +++ b/tests/test_codex_trust.py @@ -3,12 +3,11 @@ from __future__ import annotations import json -import sys from dataclasses import replace from pathlib import Path import pytest -from conftest import install_bmad_config +from conftest import install_bmad_config, write_script_launcher from bmad_loop import cli, codex_trust, probe from bmad_loop.adapters.profile import get_profile @@ -48,10 +47,10 @@ def test_scripted_app_server_executes_request_sequence_and_reads_environment(tmp """An actual zero-token child parses initialize and hooks/list, not a mocked RPC.""" data = _config(tmp_path) reply = _rpc(tmp_path, data) - script = tmp_path / "codex-stub" log = tmp_path / "requests.json" - script.write_text( - f"#!{sys.executable}\n" + script = write_script_launcher( + tmp_path, + "codex-stub", "import json, os, sys\n" "requests = []\n" "for line in sys.stdin:\n" @@ -62,9 +61,7 @@ def test_scripted_app_server_executes_request_sequence_and_reads_environment(tmp " assert os.environ['CODEX_HOME'] == 'test-home'\n" f" open({str(log)!r}, 'w').write(json.dumps(requests))\n" f" print(json.dumps({{'id': 2, 'result': {reply!r}}}), flush=True)\n", - encoding="utf-8", ) - script.chmod(0o755) profile = replace(get_profile("codex"), binary=str(script), env={"CODEX_HOME": "test-home"}) result = codex_trust.project_hook_trust(tmp_path, profile) assert result.status == "trusted", result.reason @@ -73,6 +70,27 @@ def test_scripted_app_server_executes_request_sequence_and_reads_environment(tmp assert requests[-1]["params"]["cwds"] == [str(tmp_path.resolve())] +def test_trust_resolves_codex_cmd_shim_before_spawning(tmp_path, monkeypatch): + data = _config(tmp_path) + resolved = r"C:\Program Files\nodejs\codex.cmd" + profile = replace(get_profile("codex"), env={"PATH": r"C:\Program Files\nodejs"}) + + def which(binary, *, path=None): + assert binary == "codex" + assert path == profile.env["PATH"] + return resolved + + monkeypatch.setattr(codex_trust.shutil, "which", which) + + def hooks_list(binary, cwd, env): + assert binary == resolved + assert cwd == tmp_path + return _rpc(tmp_path, data) + + monkeypatch.setattr(codex_trust, "_hooks_list", hooks_list) + assert codex_trust.project_hook_trust(tmp_path, profile).status == "trusted" + + @pytest.mark.parametrize( ("mutation", "expected"), [ @@ -118,6 +136,12 @@ def test_missing_profile_stop_and_unsupported_launch_args_fail_closed(tmp_path, codex_trust.project_hook_trust(tmp_path, replace(profile, launch_args=("-c", "x=1"))).status == "unverifiable" ) + assert ( + codex_trust.project_hook_trust( + tmp_path, replace(profile, bypass_args=("-C", "other")) + ).status + == "unverifiable" + ) assert ( codex_trust.project_hook_trust( tmp_path, replace(profile, env={"CODEX_HOME": "another"}) @@ -176,16 +200,15 @@ def test_validate_does_not_run_project_owned_codex_profile(project, tmp_path, ca policy = project.project / ".bmad-loop/policy.toml" policy.write_text('[adapter]\nname = "codex"\n', encoding="utf-8") sentinel = tmp_path / "executed" - binary = project.project / "codex-stub" - binary.write_text( - f"#!{sys.executable}\nfrom pathlib import Path\nPath({str(sentinel)!r}).write_text('yes')\n", - encoding="utf-8", + binary = write_script_launcher( + project.project, + "codex-stub", + f"from pathlib import Path\nPath({str(sentinel)!r}).write_text('yes')\n", ) - binary.chmod(0o755) overlay = project.project / ".bmad-loop/profiles/codex.toml" overlay.parent.mkdir(parents=True, exist_ok=True) overlay.write_text( - f'name = "codex"\nbinary = "{binary}"\n' + f'name = "codex"\nbinary = {json.dumps(str(binary))}\n' '[hooks]\ndialect = "codex-hooks-json"\nconfig_path = ".codex/hooks.json"\n' 'events = { SessionStart = "SessionStart", Stop = "Stop" }\n', encoding="utf-8", @@ -197,6 +220,31 @@ def test_validate_does_not_run_project_owned_codex_profile(project, tmp_path, ca assert finding["severity"] == "problem" and "project-owned" in finding["message"] +def test_validate_refuses_codex_stage_extra_args_that_change_hook_root( + project, monkeypatch, capsys +): + from bmad_loop.install import install_into + + install_bmad_config(project) + install_into(project.project, clis=("codex",)) + capsys.readouterr() + policy = project.project / ".bmad-loop/policy.toml" + policy.write_text( + '[adapter]\nname = "codex"\n[adapter.dev]\nextra_args = ["-C", "/another/project"]\n', + encoding="utf-8", + ) + monkeypatch.setattr( + codex_trust, + "project_hook_trust", + lambda *_a, **_kw: pytest.fail("a different launch root cannot certify hook trust"), + ) + cli.main(["validate", "--project", str(project.project), "--json"]) + doc = json.loads(capsys.readouterr().out) + trust = next(f for f in doc["findings"] if f["check"] == "hooks.trust") + assert trust["severity"] == "problem" + assert "adapter.extra_args" in trust["message"] and "dev" in trust["message"] + + def test_scan_and_live_probe_refuse_trust_at_their_own_directories(tmp_path, monkeypatch): profile = get_profile("codex") _config(tmp_path) @@ -321,20 +369,19 @@ def test_probe_scan_with_unregistered_codex_hooks_is_non_green(tmp_path, monkeyp def test_continuous_unrelated_messages_cannot_extend_rpc_deadline(tmp_path, monkeypatch): - script = tmp_path / "chatty-codex" - script.write_text( - f"#!{sys.executable}\n" - "import json, sys\n" + script = write_script_launcher( + tmp_path, + "chatty-codex", + "import json, sys, time\n" "for line in sys.stdin:\n" " message = json.loads(line)\n" " if message.get('method') == 'initialize':\n" " print(json.dumps({'id': 1, 'result': {}}), flush=True)\n" " if message.get('method') == 'hooks/list':\n" - " while True:\n" + " deadline = time.monotonic() + 1\n" + " while time.monotonic() < deadline:\n" " print(json.dumps({'method': 'unrelated'}), flush=True)\n", - encoding="utf-8", ) - script.chmod(0o755) monkeypatch.setattr(codex_trust, "_TIMEOUT_S", 0.1) with pytest.raises(TimeoutError): codex_trust._hooks_list(str(script), tmp_path, {}) From 225a5ca0f1b6d303dd3b07f97057b6deaf610346 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 21 Sep 2026 15:21:58 -0700 Subject: [PATCH 3/4] story 10: implemented and reviewed via bmad-loop --- src/bmad_loop/adapters/generic.py | 7 ++- src/bmad_loop/cli.py | 25 +++++++--- src/bmad_loop/codex_trust.py | 32 +++++++++--- src/bmad_loop/probe.py | 20 +++++++- tests/test_codex_trust.py | 81 +++++++++++++++++++++++++++---- tests/test_generic_tmux.py | 3 +- 6 files changed, 143 insertions(+), 25 deletions(-) diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py index e5b067fab..650cbc77d 100644 --- a/src/bmad_loop/adapters/generic.py +++ b/src/bmad_loop/adapters/generic.py @@ -726,8 +726,13 @@ def interactive_argv(self, spec: SessionSpec) -> list[str]: extra = self.extra_args if extra is None: extra = self.profile.bypass_args + binary = self.binary + if self.profile.hooks.dialect == "codex-hooks-json": + from ..codex_trust import resolved_codex_binary + + binary = resolved_codex_binary(binary, self.profile.env) or binary argv = [ - self.binary, + binary, *self.profile.launch_args, self.profile.render_prompt(spec.prompt), *extra, diff --git a/src/bmad_loop/cli.py b/src/bmad_loop/cli.py index 25cfcc06b..262649d4b 100644 --- a/src/bmad_loop/cli.py +++ b/src/bmad_loop/cli.py @@ -5052,17 +5052,22 @@ def cmd_probe(args: argparse.Namespace) -> int: ) profile = None + codex_profile_error = False try: profile = get_profile(args.cli, project) except ProfileError as e: + if args.cli == "codex": + codex_profile_error = True if not args.binary: - print(f"FAIL: {e}", file=sys.stderr) + prefix = "Codex hook trust unverifiable: " if codex_profile_error else "" + print(f"FAIL: {prefix}{e}", file=sys.stderr) return 1 # Human-facing notice — stderr in JSON mode, where stdout is the document. - print( - f" ok: unknown profile {args.cli!r}; reduced {noun} from --binary {args.binary}", - file=sys.stderr if args.json else sys.stdout, - ) + if not codex_profile_error: + print( + f" ok: unknown profile {args.cli!r}; reduced {noun} from --binary {args.binary}", + file=sys.stderr if args.json else sys.stdout, + ) if profile is not None and profile.hookless: print( @@ -5090,7 +5095,11 @@ def cmd_probe(args: argparse.Namespace) -> int: if args.probe: if profile is None: - print("FAIL: --probe needs a known profile (its hook dialect/events)", file=sys.stderr) + prefix = "Codex hook trust unverifiable: " if codex_profile_error else "" + print( + f"FAIL: {prefix}--probe needs a known profile (its hook dialect/events)", + file=sys.stderr, + ) return 1 finding = probe_mod.probe( cli=args.cli, @@ -5105,6 +5114,10 @@ def cmd_probe(args: argparse.Namespace) -> int: finding = probe_mod.scan( cli=args.cli, profile=profile, project=project, hints=hints, pseudo=pseudo ) + if codex_profile_error: + finding.hook_trust = "unverifiable" + finding.warnings.append("Codex hook trust unverifiable: profile cannot be loaded") + finding.next_steps.append("Repair the Codex profile, then re-run the probe") # One or the other, never both: --json selects the pure JSON document # (machine.py contract), otherwise the human-readable markdown report. diff --git a/src/bmad_loop/codex_trust.py b/src/bmad_loop/codex_trust.py index c93bbc5b5..13141a3e0 100644 --- a/src/bmad_loop/codex_trust.py +++ b/src/bmad_loop/codex_trust.py @@ -18,6 +18,7 @@ from pathlib import Path from .adapters.profile import CLIProfile +from .install import _hook_command from .process_host import ProcessHostError, get_process_host _EVENTS = {"SessionStart": "sessionStart", "Stop": "stop"} @@ -38,7 +39,14 @@ def hook_discovery_args_safe(args: tuple[str, ...] | None) -> bool: return args is None or all(arg == _SAFE_BYPASS_ARG for arg in args) -def _commands(config: object, profile: CLIProfile, marker: str) -> dict[str, list[str]] | None: +def resolved_codex_binary(binary: str, env: dict[str, str]) -> str | None: + """Resolve with the same PATH used by the trust query and session launch.""" + return shutil.which(binary, path={**os.environ, **env}.get("PATH")) + + +def _commands( + config: object, profile: CLIProfile, project: Path, marker: str +) -> dict[str, list[str]] | None: if not isinstance(config, dict) or not isinstance(config.get("hooks"), dict): raise ValueError("malformed Codex hook config") events = profile.hooks.events @@ -46,6 +54,13 @@ def _commands(config: object, profile: CLIProfile, marker: str) -> dict[str, lis return None found: dict[str, list[str]] = {} for canonical in _EVENTS: + if marker == _RELAY_MARKER: + expected_command = _hook_command(project, profile, canonical) + else: + host = get_process_host() + expected_command = ( + f"{host.hook_interpreter()} {host.shell_quote(str(project / marker))} {canonical}" + ) handlers = config["hooks"].get(canonical) if handlers is None: return None @@ -60,6 +75,13 @@ def _commands(config: object, profile: CLIProfile, marker: str) -> dict[str, lis raise ValueError("malformed Codex hook entry") command = hook.get("command") if isinstance(command, str) and marker in command: + # A SessionStart matcher can exclude startup even when + # Codex reports the command trusted and enabled. The + # installed relay has none; refuse customized matchers. + if canonical == "SessionStart" and group.get("matcher") not in (None, ""): + return None + if command != expected_command: + return None commands.append(command) if not commands: return None @@ -177,18 +199,16 @@ def project_hook_trust( except (OSError, UnicodeError, ValueError): return TrustResult("unverifiable", "hook trust config is unreadable") try: - commands = _commands(config, profile, marker) + commands = _commands(config, profile, project, marker) except ValueError: return TrustResult("unverifiable", "hook trust config has malformed fields") if commands is None: return TrustResult( - "untrusted", "hook trust: a required SessionStart or Stop hook is not registered" + "untrusted", "hook trust: a required SessionStart or Stop relay is not usable" ) # Windows npm installs expose a codex.cmd shim through PATHEXT. Popen with # a bare name need not find it; which() returns the executable run would use. - resolved_binary = shutil.which( - binary or profile.binary, path={**os.environ, **profile.env}.get("PATH") - ) + resolved_binary = resolved_codex_binary(binary or profile.binary, profile.env) if resolved_binary is None: return TrustResult("unverifiable", "hook trust Codex binary is unavailable") try: diff --git a/src/bmad_loop/probe.py b/src/bmad_loop/probe.py index 4181e42d3..e232fb170 100644 --- a/src/bmad_loop/probe.py +++ b/src/bmad_loop/probe.py @@ -494,10 +494,16 @@ def _check_hook_trust( "Codex has no trust grant for this fresh temporary workspace; " "live capture cannot proceed until that workspace is trusted" ) - else: + elif "stale" in trust.reason: finding.next_steps.append( "Open Codex in the project checkout and accept its hook trust prompt" ) + elif "not registered" in trust.reason or "omitted" in trust.reason: + finding.next_steps.append( + f"Re-register the Codex relay with `bmad-loop init --cli {profile.name}`" + ) + else: + finding.next_steps.append("Resolve the Codex hook trust diagnostic and re-run the scan") # ----------------------------------------------------------------- SCAN mode @@ -602,6 +608,10 @@ def kill(self) -> None: def _probe_argv(profile: CLIProfile, binary: str, hints: Hints) -> list[str]: + if profile.hooks.dialect == "codex-hooks-json": + from .codex_trust import resolved_codex_binary + + binary = resolved_codex_binary(binary, profile.env) or binary argv = [ binary, *profile.launch_args, @@ -706,7 +716,13 @@ def probe( mux_ready = bool(mux.available()) except Exception: # a raising host probe means "cannot probe", not a crash mux_ready = False - if not mux_ready or not shutil.which(binary): + if profile.hooks.dialect == "codex-hooks-json": + from .codex_trust import resolved_codex_binary + + binary_found = resolved_codex_binary(binary, profile.env) is not None + else: + binary_found = shutil.which(binary) is not None + if not mux_ready or not binary_found: # finding.binary, not the raw local — see the identical note in scan() missing = f"multiplexer backend {type(mux).__name__}" if not mux_ready else finding.binary finding.warnings.append(f"{missing} not on PATH — cannot probe; falling back to scan") diff --git a/tests/test_codex_trust.py b/tests/test_codex_trust.py index a678fd611..2c43c9b62 100644 --- a/tests/test_codex_trust.py +++ b/tests/test_codex_trust.py @@ -11,15 +11,14 @@ from bmad_loop import cli, codex_trust, probe from bmad_loop.adapters.profile import get_profile -from bmad_loop.install import merge_hooks +from bmad_loop.install import _hook_command, merge_hooks def _config(root: Path, commands: dict[str, str] | None = None) -> dict: profile = get_profile("codex") if commands is None: commands = { - event: f"python3 {root}/.bmad-loop/bmad_loop_hook.py {event}" - for event in ("SessionStart", "Stop") + event: _hook_command(root, profile, event) for event in ("SessionStart", "Stop") } data, _ = merge_hooks({}, commands, profile.hooks.dialect) path = root / profile.hooks.config_path @@ -95,8 +94,10 @@ def hooks_list(binary, cwd, env): ("mutation", "expected"), [ ("start-modified", "untrusted"), + ("start-untrusted", "untrusted"), ("start-omitted", "untrusted"), ("modified", "untrusted"), + ("untrusted", "untrusted"), ("disabled", "untrusted"), ("omitted", "untrusted"), ("wrong-command", "untrusted"), @@ -110,10 +111,14 @@ def test_trust_refuses_stale_or_unmatched_relay(tmp_path, monkeypatch, mutation, hooks = result["data"][0]["hooks"] if mutation == "start-modified": hooks[0]["trustStatus"] = "modified" + elif mutation == "start-untrusted": + hooks[0]["trustStatus"] = "untrusted" elif mutation == "start-omitted": hooks.pop(0) elif mutation == "modified": hooks[1]["trustStatus"] = "modified" + elif mutation == "untrusted": + hooks[1]["trustStatus"] = "untrusted" elif mutation == "disabled": hooks[1]["enabled"] = False elif mutation == "omitted": @@ -128,6 +133,23 @@ def test_trust_refuses_stale_or_unmatched_relay(tmp_path, monkeypatch, mutation, assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == expected +def test_trust_refuses_relay_for_old_checkout_and_startup_excluding_matcher(tmp_path, monkeypatch): + data = _config(tmp_path) + monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: _rpc(tmp_path, data)) + assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == "trusted" + + data["hooks"]["SessionStart"][0]["matcher"] = "^resume$" + (tmp_path / ".codex/hooks.json").write_text(json.dumps(data), encoding="utf-8") + assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == "untrusted" + + data["hooks"]["SessionStart"][0].pop("matcher") + data["hooks"]["Stop"][0]["hooks"][0]["command"] = _hook_command( + tmp_path / "old-checkout", get_profile("codex"), "Stop" + ) + (tmp_path / ".codex/hooks.json").write_text(json.dumps(data), encoding="utf-8") + assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == "untrusted" + + def test_missing_profile_stop_and_unsupported_launch_args_fail_closed(tmp_path, monkeypatch): data = _config(tmp_path) monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: _rpc(tmp_path, data)) @@ -191,6 +213,31 @@ def test_validate_names_untrusted_hook_and_refuses_worktree_inference(project, m assert len(findings) == 1 and "worktree" in findings[0]["message"] +def test_validate_trust_query_uses_selected_project(project, tmp_path, monkeypatch, capsys): + from bmad_loop.install import install_into + + install_bmad_config(project) + install_into(project.project, clis=("codex",)) + capsys.readouterr() + (project.project / ".bmad-loop/policy.toml").write_text( + '[adapter]\nname = "codex"\n', encoding="utf-8" + ) + monkeypatch.chdir(tmp_path) + queried = [] + + def trust(path, _profile): + queried.append(path) + return codex_trust.TrustResult("trusted", "hook trust current") + + monkeypatch.setattr(codex_trust, "project_hook_trust", trust) + cli.main(["validate", "--project", str(project.project), "--json"]) + finding = next( + f for f in json.loads(capsys.readouterr().out)["findings"] if f["check"] == "hooks.trust" + ) + assert finding["severity"] == "ok" + assert queried == [project.project] + + def test_validate_does_not_run_project_owned_codex_profile(project, tmp_path, capsys): from bmad_loop.install import install_into @@ -220,8 +267,9 @@ def test_validate_does_not_run_project_owned_codex_profile(project, tmp_path, ca assert finding["severity"] == "problem" and "project-owned" in finding["message"] +@pytest.mark.parametrize("role", ["dev", "review", "triage"]) def test_validate_refuses_codex_stage_extra_args_that_change_hook_root( - project, monkeypatch, capsys + project, monkeypatch, capsys, role ): from bmad_loop.install import install_into @@ -230,7 +278,8 @@ def test_validate_refuses_codex_stage_extra_args_that_change_hook_root( capsys.readouterr() policy = project.project / ".bmad-loop/policy.toml" policy.write_text( - '[adapter]\nname = "codex"\n[adapter.dev]\nextra_args = ["-C", "/another/project"]\n', + f'[adapter]\nname = "codex"\n[adapter.{role}]\n' + 'extra_args = ["-C", "/another/project"]\n', encoding="utf-8", ) monkeypatch.setattr( @@ -242,7 +291,7 @@ def test_validate_refuses_codex_stage_extra_args_that_change_hook_root( doc = json.loads(capsys.readouterr().out) trust = next(f for f in doc["findings"] if f["check"] == "hooks.trust") assert trust["severity"] == "problem" - assert "adapter.extra_args" in trust["message"] and "dev" in trust["message"] + assert "adapter.extra_args" in trust["message"] and role in trust["message"] def test_scan_and_live_probe_refuse_trust_at_their_own_directories(tmp_path, monkeypatch): @@ -277,7 +326,7 @@ def kill(self): monkeypatch.setattr(probe, "get_multiplexer", Mux) monkeypatch.setattr(probe, "_ProbeLauncher", Launcher) - monkeypatch.setattr(probe.shutil, "which", lambda _binary: "/bin/true") + monkeypatch.setattr(probe.shutil, "which", lambda _binary, **_kwargs: "/bin/true") live = probe.probe( cli="codex", profile=profile, project=tmp_path, hints=probe.Hints(binary="chosen") ) @@ -323,7 +372,7 @@ def wait_for(self, *_args, **_kwargs): monkeypatch.setattr(probe, "get_multiplexer", Mux) monkeypatch.setattr(probe, "_ProbeLauncher", Launcher) monkeypatch.setattr(probe, "SignalWatcher", Watcher) - monkeypatch.setattr(probe.shutil, "which", lambda _binary: "/bin/true") + monkeypatch.setattr(probe.shutil, "which", lambda _binary, **_kwargs: "/bin/true") monkeypatch.setattr(probe, "run_version_help", lambda binary: probe.FlagFinding(binary, True)) monkeypatch.setattr(probe, "discover_transcript", lambda *_args, **_kwargs: None) monkeypatch.setattr(probe.time, "sleep", lambda _seconds: None) @@ -334,7 +383,7 @@ def wait_for(self, *_args, **_kwargs): assert finding.hook_trust == "trusted" assert events[0][0] == "trust" and events[0][1] != tmp_path assert events[0][2:] == ("chosen", probe.PROBE_HOOK_NAME) - assert events[1] == ("start", events[0][1], "chosen") + assert events[1] == ("start", events[0][1], "/bin/true") assert events[-1] == ("kill",) @@ -368,6 +417,20 @@ def test_probe_scan_with_unregistered_codex_hooks_is_non_green(tmp_path, monkeyp assert any("hook trust" in warning for warning in doc["warnings"]) +def test_probe_codex_with_invalid_profile_cannot_fall_back_to_green_scan(tmp_path, capsys): + overlay = tmp_path / ".bmad-loop/profiles/codex.toml" + overlay.parent.mkdir(parents=True) + overlay.write_text("invalid = [", encoding="utf-8") + rc = cli.main( + ["probe-adapter", "codex", "--project", str(tmp_path), "--binary", "chosen", "--json"] + ) + out, err = capsys.readouterr() + doc = json.loads(out) + assert rc == 1 and "FAIL" in err + assert doc["hook_trust"] == "unverifiable" + assert any("hook trust unverifiable" in warning for warning in doc["warnings"]) + + def test_continuous_unrelated_messages_cannot_extend_rpc_deadline(tmp_path, monkeypatch): script = write_script_launcher( tmp_path, diff --git a/tests/test_generic_tmux.py b/tests/test_generic_tmux.py index 9781657be..9ead259c7 100644 --- a/tests/test_generic_tmux.py +++ b/tests/test_generic_tmux.py @@ -170,8 +170,9 @@ def test_build_command_claude(tmp_path): def test_build_command_codex_renders_skill_mention(tmp_path): adapter = make_adapter(tmp_path, profile_name="codex") cmd = adapter.build_command(make_spec(tmp_path)) + binary = shutil.which("codex") or "codex" assert cmd.startswith( - "codex 'Use the $bmad-dev-auto skill now, and use subagents as needed: 1-1-a'" + f"{binary} 'Use the $bmad-dev-auto skill now, and use subagents as needed: 1-1-a'" ) assert "--dangerously-bypass-approvals-and-sandbox" in cmd assert cmd.endswith("--model sonnet") From 9ea2dcc84a2b18ce548115ebc8ba5dfcde6fb6e7 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 21 Sep 2026 15:29:48 -0700 Subject: [PATCH 4/4] Make Codex trust tests independent of local CLI installation --- tests/test_codex_trust.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_codex_trust.py b/tests/test_codex_trust.py index 2c43c9b62..75ea594da 100644 --- a/tests/test_codex_trust.py +++ b/tests/test_codex_trust.py @@ -107,6 +107,7 @@ def hooks_list(binary, cwd, env): ) def test_trust_refuses_stale_or_unmatched_relay(tmp_path, monkeypatch, mutation, expected): data = _config(tmp_path) + monkeypatch.setattr(codex_trust, "resolved_codex_binary", lambda *_: "codex-stub") result = _rpc(tmp_path, data) hooks = result["data"][0]["hooks"] if mutation == "start-modified": @@ -135,6 +136,7 @@ def test_trust_refuses_stale_or_unmatched_relay(tmp_path, monkeypatch, mutation, def test_trust_refuses_relay_for_old_checkout_and_startup_excluding_matcher(tmp_path, monkeypatch): data = _config(tmp_path) + monkeypatch.setattr(codex_trust, "resolved_codex_binary", lambda *_: "codex-stub") monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: _rpc(tmp_path, data)) assert codex_trust.project_hook_trust(tmp_path, get_profile("codex")).status == "trusted" @@ -152,6 +154,7 @@ def test_trust_refuses_relay_for_old_checkout_and_startup_excluding_matcher(tmp_ def test_missing_profile_stop_and_unsupported_launch_args_fail_closed(tmp_path, monkeypatch): data = _config(tmp_path) + monkeypatch.setattr(codex_trust, "resolved_codex_binary", lambda *_: "codex-stub") monkeypatch.setattr(codex_trust, "_hooks_list", lambda *_: _rpc(tmp_path, data)) profile = get_profile("codex") assert (