diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 5c38ded..6dbc9a9 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -6,5 +6,19 @@ "homepage": "https://github.com/SollanSystems/loop-engineer", "repository": "https://github.com/SollanSystems/loop-engineer", "keywords": ["agent", "loop", "agentic", "verification", "harness", "orchestration", "self-improvement", "claude-code", "eval", "repair"], - "license": "MIT" + "license": "MIT", + "hooks": { + "Stop": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "python3 ${CLAUDE_PLUGIN_ROOT}/hooks/stop_firewall.py", + "timeout": 90 + } + ] + } + ] + } } diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d86bf5..6116702 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,28 @@ same evidence cross-check `loop doctor` enforces, applied before the file exists the package stays zero-dependency; a dedicated `recipe (langgraph)` CI job installs LangGraph and runs it. +**A1 — the Stop-hook firewall.** The false-completion wedge, enforced at the +session boundary instead of only on demand. When a `.loop/` contract claims +`Succeeded` while `loop doctor` still reports `ok:false`, the Stop hook blocks the +turn from ending and hands the agent the named doctor issues, so a run cannot exit +on a false "done". It is fail-open by construction — a broken or unresolvable +firewall never locks a session — and a strict no-op for every repo without a +`.loop/` contract. + +### Added +- **`hooks/stop_firewall.py`** — a stdlib-only Stop hook that blocks a + `Succeeded`-claiming contract whose `loop doctor` report is `ok:false`, carrying + the issues into the block reason. Fails open on any error (malformed stdin, + unresolvable `loop` CLI, doctor failure), stays silent when no `.loop/` exists, + respects `stop_hook_active` to avoid livelock, and blocks at most once per + session per issue-set (a tempdir sentinel keyed on the issue digest). Covered by + `scripts/test_stop_firewall.py` (subprocess acceptance tests for the honest, + lying, in-flight, absent, once-per-session, and fail-open paths). +- **Plugin-manifest registration** — the hook is wired into + `.claude-plugin/plugin.json` under the top-level `hooks.Stop` key + (`python3 ${CLAUDE_PLUGIN_ROOT}/hooks/stop_firewall.py`), so a marketplace + install gets the firewall with zero configuration. + ## 0.6.1 — 2026-07-04 **PyPI substrate.** `loop-engineer` becomes a self-contained wheel that runs from diff --git a/hooks/stop_firewall.py b/hooks/stop_firewall.py new file mode 100644 index 0000000..de58900 --- /dev/null +++ b/hooks/stop_firewall.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""Stop-hook false-completion firewall (A1). + +On session stop, if the CWD holds a .loop/ contract that claims Succeeded while +`loop doctor` reports ok:false, emit blocking feedback carrying the doctor +issues so the agent cannot end the turn on a false "done". + +Invariants: + * strict no-op when no .loop/ exists — zero cost for every other repo; + * fail-open on ANY error — a broken firewall must never lock a session; + * blocks at most once per session per issue-set (tempdir sentinel), and never + when stop_hook_active is set — no livelock. + +Stdlib only. Runs under whatever python3 Claude Code invokes. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +_MAX_ISSUES_IN_REASON = 5 + + +def _cli_command() -> list[str] | None: + exe = shutil.which("loop") + if exe: + return [exe] + root = os.environ.get("CLAUDE_PLUGIN_ROOT", "") + if root and (Path(root) / "loop" / "__main__.py").is_file(): + return [sys.executable or "python3", "-m", "loop"] + return None + + +def _cli_env() -> dict[str, str]: + env = dict(os.environ) + root = env.get("CLAUDE_PLUGIN_ROOT", "") + if root: + env["PYTHONPATH"] = root + os.pathsep + env.get("PYTHONPATH", "") + return env + + +def _claims_succeeded(loop_dir: Path) -> bool: + for candidate, key in ( + (loop_dir / "terminal_state.json", "state"), + (loop_dir / "state.json", "terminal_state"), + ): + try: + data = json.loads(candidate.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + if isinstance(data, dict) and data.get(key) == "Succeeded": + return True + return False + + +def _blocked_before(session_id: str, digest: str) -> bool: + sentinel = Path(tempfile.gettempdir()) / f"loop-engineer-stop-{session_id or 'nosession'}" + try: + if sentinel.is_file() and sentinel.read_text(encoding="utf-8") == digest: + return True + sentinel.write_text(digest, encoding="utf-8") + except OSError: + return True # cannot track repeats → err on the never-lock side + return False + + +def main() -> int: + payload = json.load(sys.stdin) + cwd = Path(payload.get("cwd") or os.getcwd()) + loop_dir = cwd / ".loop" + if not loop_dir.is_dir(): + return 0 + if payload.get("stop_hook_active"): + return 0 + if not _claims_succeeded(loop_dir): + return 0 + + cli = _cli_command() + if cli is None: + return 0 + proc = subprocess.run( + cli + ["doctor", str(cwd)], + capture_output=True, text=True, timeout=60, env=_cli_env(), # < manifest's 90s hook timeout, so this dies first + ) + report = json.loads(proc.stdout) + if report.get("ok") is True: + return 0 + + issues = [i for i in report.get("issues", []) if isinstance(i, dict)] + digest = hashlib.sha256(json.dumps(issues, sort_keys=True).encode("utf-8")).hexdigest() + if _blocked_before(str(payload.get("session_id", "")), digest): + return 0 + + summary = "; ".join( + f"{i.get('code', '?')}: {i.get('message', '')}" for i in issues[:_MAX_ISSUES_IN_REASON] + ) or "doctor reported ok:false" + if len(issues) > _MAX_ISSUES_IN_REASON: + summary += f"; … {len(issues) - _MAX_ISSUES_IN_REASON} more" + print(json.dumps({ + "decision": "block", + "reason": ( + "loop-engineer stop firewall: this workspace's loop contract claims " + f"Succeeded, but `loop doctor` reports {len(issues)} issue(s): {summary}. " + "Fix the contract or record an honest terminal state " + "(e.g. FailedUnverifiable) before ending the turn." + ), + })) + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Exception: + sys.exit(0) # fail-open, always diff --git a/scripts/test_stop_firewall.py b/scripts/test_stop_firewall.py new file mode 100644 index 0000000..a138072 --- /dev/null +++ b/scripts/test_stop_firewall.py @@ -0,0 +1,156 @@ +"""A1 acceptance, exercised offline: honest contract passes silently, lying +contract blocks with the doctor issues named, absent .loop is a strict no-op, +and any error path fails OPEN (a broken firewall must never lock a session).""" + +from __future__ import annotations + +import json +import subprocess +import sys +import uuid +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +HOOK = REPO_ROOT / "hooks" / "stop_firewall.py" + +sys.path.insert(0, str(REPO_ROOT)) +from loop.scaffold import scaffold # noqa: E402 + + +def _run_hook(payload: dict, plugin_root: Path | None = REPO_ROOT, extra_env: dict | None = None): + env = { + "PATH": "/usr/bin:/bin", # no `loop` console script on PATH: forces the plugin-root path + "CLAUDE_PLUGIN_ROOT": str(plugin_root) if plugin_root else "", + } + if extra_env: + env.update(extra_env) + return subprocess.run( + [sys.executable, "-B", str(HOOK)], + input=json.dumps(payload), env=env, capture_output=True, text=True, timeout=120, + ) + + +def _payload(cwd: Path, **overrides) -> dict: + base = { + "session_id": f"test-{uuid.uuid4().hex}", # unique: the once-per-session sentinel must not leak across tests + "transcript_path": "/dev/null", + "cwd": str(cwd), + "hook_event_name": "Stop", + } + base.update(overrides) + return base + + +def _lying_workspace(tmp_path: Path) -> Path: + ws = tmp_path / "lying" + scaffold(ws) + (ws / ".loop" / "terminal_state.json").write_text(json.dumps({ + "schema": "loop-engineer/terminal@1", + "state": "Succeeded", + "criteria_met": {"1": False}, # no met criterion → doctor ok:false (G1) + "evidence": [], + "false_completion": True, # contradiction → doctor ok:false + }), encoding="utf-8") + return ws + + +def _honest_workspace(tmp_path: Path) -> Path: + ws = tmp_path / "honest" + scaffold(ws) + (ws / ".loop" / "terminal_state.json").write_text(json.dumps({ + "schema": "loop-engineer/terminal@1", + "state": "Succeeded", + "criteria_met": {"1": True}, + "evidence": ["artifact.txt"], + "false_completion": False, + }), encoding="utf-8") + state_path = ws / ".loop" / "state.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + state["terminal_state"] = "Succeeded" + state_path.write_text(json.dumps(state, indent=2), encoding="utf-8") + return ws + + +def test_no_loop_dir_is_a_strict_noop(tmp_path): + proc = _run_hook(_payload(tmp_path)) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_honest_succeeded_passes_silently(tmp_path): + proc = _run_hook(_payload(_honest_workspace(tmp_path))) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_inflight_contract_passes_silently(tmp_path): + ws = tmp_path / "inflight" + scaffold(ws) # no terminal claim at all + proc = _run_hook(_payload(ws)) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_lying_succeeded_blocks_with_doctor_issues(tmp_path): + proc = _run_hook(_payload(_lying_workspace(tmp_path)), extra_env={"TMPDIR": str(tmp_path)}) + assert proc.returncode == 0 + out = json.loads(proc.stdout) + assert out["decision"] == "block" + assert "contradictory_terminal" in out["reason"] + assert "Succeeded" in out["reason"] + + +def test_blocks_at_most_once_per_session(tmp_path): + ws = _lying_workspace(tmp_path) + payload = _payload(ws) + extra_env = {"TMPDIR": str(tmp_path)} # shared across both calls: the sentinel must persist between them + first = _run_hook(payload, extra_env=extra_env) + assert json.loads(first.stdout)["decision"] == "block" + second = _run_hook(payload, extra_env=extra_env) # same session_id, same issues + assert second.returncode == 0 + assert second.stdout.strip() == "" + + +def test_stop_hook_active_never_blocks(tmp_path): + proc = _run_hook(_payload(_lying_workspace(tmp_path), stop_hook_active=True)) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_malformed_stdin_fails_open(): + proc = subprocess.run( + [sys.executable, "-B", str(HOOK)], input="not json{{{", + env={"PATH": "/usr/bin:/bin", "CLAUDE_PLUGIN_ROOT": str(REPO_ROOT)}, + capture_output=True, text=True, timeout=120, + ) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_unresolvable_cli_fails_open(tmp_path): + """Forced-error fixture: lying contract but no reachable loop CLI.""" + proc = _run_hook(_payload(_lying_workspace(tmp_path)), plugin_root=tmp_path / "empty") + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_doctor_garbage_output_fails_open(tmp_path): + """Forced-error fixture: a resolvable loop CLI that prints non-JSON to stdout.""" + fake_bin = tmp_path / "fakebin" + fake_bin.mkdir() + fake_loop = fake_bin / "loop" + fake_loop.write_text('#!/bin/sh\necho "not json"\nexit 0\n', encoding="utf-8") + fake_loop.chmod(0o755) + proc = _run_hook( + _payload(_lying_workspace(tmp_path)), + extra_env={"PATH": f"{fake_bin}:/usr/bin:/bin"}, # fake `loop` wins shutil.which() over the real one + ) + assert proc.returncode == 0 + assert proc.stdout.strip() == "" + + +def test_hook_is_registered_in_plugin_manifest(): + manifest = json.loads((REPO_ROOT / ".claude-plugin" / "plugin.json").read_text(encoding="utf-8")) + stop_entries = manifest["hooks"]["Stop"] + commands = [h["command"] for entry in stop_entries for h in entry["hooks"]] + assert any("stop_firewall.py" in c and "${CLAUDE_PLUGIN_ROOT}" in c for c in commands)