feat(a1): Stop-hook false-completion firewall

.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
+          }
+        ]
+      }
+    ]
+  }
 }

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

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

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)