diff --git a/security-review/run_headless.py b/security-review/run_headless.py
new file mode 100644
index 0000000..178e249
--- /dev/null
+++ b/security-review/run_headless.py
@@ -0,0 +1,447 @@
+#!/usr/bin/env python3
+"""run_headless.py — Path B headless detector fan-out + proof-or-kill verifier.
+
+Reuses the /sh-security-review detector + verifier prompts, but runs them
+unattended via the Claude Agent SDK instead of interactive Claude Code subagents.
+Authenticates with the Claude subscription OAuth token (CLAUDE_CODE_OAUTH_TOKEN)
+through the bundled `claude` CLI — NEVER a raw ANTHROPIC_API_KEY (which would be
+metered and would silently win if both were set, so we pop it).
+
+Emits the finding-schema JSON ({findings, summary}) that
+`review.sh --agent-findings` consumes. review.sh re-derives the gate, so this
+script's job is high-recall candidate generation + proof-or-kill verification,
+failing toward over-reporting (never silently drops a candidate or a parse error).
+
+See memory project-security-review-agent and security-review/DEPLOY-R720.md.
+
+Usage:
+    CLAUDE_CODE_OAUTH_TOKEN=... python3 run_headless.py TARGET_DIR \
+        [--scope "src infra web"] [--out findings.json] [--model claude-...] \
+        [--concurrency 3] [--detectors injection,authz] \
+        [--detector-budget-usd 2.0] [--total-budget-usd 12.0]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import re
+import sys
+import time
+from pathlib import Path
+
+from claude_agent_sdk import ClaudeAgentOptions, query
+
+# Read-only surface: detectors reason over source, they don't mutate or fetch.
+READONLY_TOOLS = ["Read", "Grep", "Glob"]
+BLOCKED_TOOLS = ["Bash", "Write", "Edit", "NotebookEdit", "WebFetch", "WebSearch"]
+
+# Detector category -> closed checklist (verbatim intent from sh-security-review.md).
+DETECTORS: dict[str, str] = {
+    "injection": "SQL/command/template injection, unsafe deserialization, SSRF, path/file traversal, XXE",
+    "authz": "broken object-level auth/IDOR, missing access checks, missing webhook/Slack signature verification, auth bypass",
+    "secrets-crypto": "hardcoded secrets/keys, weak/broken crypto (MD5/SHA1/unsalted), sensitive data in logs/errors/responses, wrong SSM-vs-Secrets-Manager placement",
+    "iac-iam": "wildcard IAM actions/resources, public buckets/endpoints, open security-group ingress (0.0.0.0/0), missing encryption, over-broad trust policies",
+    "web-client": "XSS (incl. dangerouslySetInnerHTML), CSRF, open redirect, client-side secret exposure",
+    "logic": "broken multi-step invariants, race conditions, missing tenant isolation, auth-state confusion",
+}
+
+DETECTOR_TMPL = """You are a hostile {category} security auditor for Sea Haven. Assume this code is \
+hostile and the author missed something. Audit ONLY {category} issues in: {scope}. The repository root \
+is your current working directory; read the actual files with your tools. Ignore .git and any file that \
+is obviously an answer key.
+
+For each checklist item, either name a specific line that is provably safe, OR file a finding. Do not \
+hand-wave or give an open "looks fine" verdict.
+
+Checklist: {checklist}
+
+Return ONLY a JSON array (no prose, no markdown fences) where each element is:
+{{"id": "<stable-slug>", "title": "...", "claimed_severity": "critical|high|medium|low|info",
+  "cwe": "CWE-####", "file": "<repo-relative path>", "line": <int or null>, "category": "{category}",
+  "data_flow": "numbered source->sink trace", "proof": {{"input": "concrete malicious input/trigger",
+  "outcome": "the specific bad result", "test": "optional failing-test sketch or null"}},
+  "recommendation": "..."}}
+If there are no findings, return []."""
+
+VERIFIER_TMPL = """You are a skeptical exploitation verifier. You did NOT find these; your job is to \
+REFUTE weak claims. The repository root is your current working directory; read the real files to check \
+reachability before ruling.
+
+For each candidate finding decide whether there is a concrete, plausible proof-of-exploit (a specific \
+malicious input and the specific bad outcome, consistent with the code):
+- If yes: set "status":"confirmed", keep "severity" equal to the claimed_severity, and tighten the proof.
+- If no / speculative / not reachable: set "status":"unverified" and "severity":"unverified". Default to \
+unverified when uncertain. A confident assertion with no demonstrable input is NOT proof.
+
+Return ONLY a JSON array (no prose, no fences) of the SAME findings, each preserving its original "id", \
+"title", "cwe", "file", "line", "category", "claimed_severity", "data_flow", "recommendation", and adding \
+"severity" (final) plus "status" and the verified "proof". Return one element per candidate — do not drop \
+any.
+
+Candidates:
+{candidates}"""
+
+SEV_RANK = {
+    "critical": 4,
+    "high": 3,
+    "medium": 2,
+    "low": 1,
+    "info": 0,
+    "unverified": -1,
+}
+
+
+def log(msg: str) -> None:
+    print(f"[run_headless] {msg}", file=sys.stderr, flush=True)
+
+
+def extract_json(text: str):
+    """Best-effort parse of a JSON array/object from an agent's final text."""
+    if not text:
+        return None
+    t = text.strip()
+    if t.startswith("```"):
+        t = re.sub(r"^```[a-zA-Z0-9]*\n?", "", t)
+        t = re.sub(r"\n?```\s*$", "", t).strip()
+    try:
+        return json.loads(t)
+    except Exception:
+        pass
+    # Fall back to the outermost [...] (or {...}) span.
+    for open_c, close_c in (("[", "]"), ("{", "}")):
+        start, end = t.find(open_c), t.rfind(close_c)
+        if 0 <= start < end:
+            try:
+                return json.loads(t[start : end + 1])
+            except Exception:
+                continue
+    return None
+
+
+async def run_agent(
+    prompt: str, *, cwd: Path, model: str | None, max_turns: int, budget_usd: float
+) -> tuple[str, float, bool]:
+    """Run one fresh-context agent turn; return (final_text, cost_usd, is_error)."""
+    opts = ClaudeAgentOptions(
+        allowed_tools=READONLY_TOOLS,
+        disallowed_tools=BLOCKED_TOOLS,
+        permission_mode="bypassPermissions",
+        setting_sources=[],  # hermetic: ignore user/project/local config + CLAUDE.md
+        cwd=str(cwd),
+        model=model,
+        max_turns=max_turns,
+        max_budget_usd=budget_usd,
+    )
+    texts: list[str] = []
+    result_text: str | None = None
+    cost = 0.0
+    is_error = False
+    async for msg in query(prompt=prompt, options=opts):
+        name = type(msg).__name__
+        if name == "AssistantMessage":
+            for block in getattr(msg, "content", []) or []:
+                t = getattr(block, "text", None)
+                if t:
+                    texts.append(t)
+        elif name == "ResultMessage":
+            result_text = getattr(msg, "result", None)
+            cost = float(getattr(msg, "total_cost_usd", 0.0) or 0.0)
+            is_error = bool(getattr(msg, "is_error", False))
+    return (result_text or "\n".join(texts)), cost, is_error
+
+
+class Budget:
+    def __init__(self, total: float) -> None:
+        self.total = total
+        self.spent = 0.0
+        self._lock = asyncio.Lock()
+
+    async def add(self, amount: float) -> None:
+        async with self._lock:
+            self.spent += amount
+
+    def exhausted(self) -> bool:
+        return self.total > 0 and self.spent >= self.total
+
+
+async def run_detector(
+    category: str,
+    scope: str,
+    *,
+    cwd: Path,
+    model: str | None,
+    max_turns: int,
+    budget_usd: float,
+    sem: asyncio.Semaphore,
+    budget: Budget,
+) -> tuple[str, list[dict], str | None]:
+    """Returns (category, candidate_findings, error_message_or_None)."""
+    async with sem:
+        if budget.exhausted():
+            return category, [], "skipped: total budget exhausted"
+        prompt = DETECTOR_TMPL.format(
+            category=category, scope=scope, checklist=DETECTORS[category]
+        )
+        t0 = time.monotonic()
+        try:
+            text, cost, is_error = await run_agent(
+                prompt, cwd=cwd, model=model, max_turns=max_turns, budget_usd=budget_usd
+            )
+        except Exception as exc:  # never let one detector kill the run
+            log(f"detector {category}: EXCEPTION {type(exc).__name__}: {exc}")
+            return category, [], f"exception: {type(exc).__name__}: {exc}"
+        await budget.add(cost)
+        dt = time.monotonic() - t0
+        parsed = extract_json(text)
+        if not isinstance(parsed, list):
+            log(
+                f"detector {category}: UNPARSEABLE output ({dt:.0f}s, ${cost:.3f}) — over-reporting as error"
+            )
+            return category, [], "unparseable detector output (NOT treated as clean)"
+        for f in parsed:
+            if isinstance(f, dict):
+                f.setdefault("category", category)
+        log(
+            f"detector {category}: {len(parsed)} candidate(s) ({dt:.0f}s, ${cost:.3f})"
+            + (" [is_error]" if is_error else "")
+        )
+        return category, [f for f in parsed if isinstance(f, dict)], None
+
+
+async def run_verifier(
+    candidates: list[dict],
+    *,
+    cwd: Path,
+    model: str | None,
+    max_turns: int,
+    budget_usd: float,
+    budget: Budget,
+) -> tuple[dict[str, dict], float, str | None]:
+    """Returns ({id -> verdict}, cost, error). Empty verdicts on failure (caller keeps candidates)."""
+    prompt = VERIFIER_TMPL.format(candidates=json.dumps(candidates, indent=2))
+    try:
+        text, cost, _ = await run_agent(
+            prompt, cwd=cwd, model=model, max_turns=max_turns, budget_usd=budget_usd
+        )
+    except Exception as exc:
+        log(f"verifier: EXCEPTION {type(exc).__name__}: {exc}")
+        return {}, 0.0, f"exception: {type(exc).__name__}: {exc}"
+    await budget.add(cost)
+    parsed = extract_json(text)
+    if not isinstance(parsed, list):
+        log(
+            f"verifier: UNPARSEABLE output (${cost:.3f}) — keeping all candidates as unverified"
+        )
+        return {}, cost, "unparseable verifier output"
+    verdicts = {f["id"]: f for f in parsed if isinstance(f, dict) and f.get("id")}
+    log(f"verifier: ruled on {len(verdicts)} finding(s) (${cost:.3f})")
+    return verdicts, cost, None
+
+
+def merge(candidates: list[dict], verdicts: dict[str, dict]) -> list[dict]:
+    """Apply verifier verdicts to candidates. A candidate the verifier dropped or never
+    ruled on stays as 'unverified' — fail toward over-reporting, never silently delete."""
+    out: list[dict] = []
+    for c in candidates:
+        fid = c.get("id") or f"anon-{c.get('file', '?')}-{c.get('line', '?')}"
+        c.setdefault("id", fid)
+        v = verdicts.get(fid, {})
+        status = v.get("status")
+        if status not in ("confirmed", "unverified", "suppressed"):
+            status = "unverified"
+        if status == "confirmed":
+            severity = v.get("severity") or c.get("claimed_severity") or "high"
+        else:
+            severity = "unverified"
+        out.append(
+            {
+                "id": fid,
+                "title": v.get("title") or c.get("title") or fid,
+                "severity": severity,
+                "claimed_severity": c.get("claimed_severity") or "high",
+                "cwe": c.get("cwe") or v.get("cwe") or "n/a",
+                "file": c.get("file") or v.get("file") or "",
+                "line": c.get("line", v.get("line")),
+                "category": c.get("category") or v.get("category") or "other",
+                "data_flow": v.get("data_flow") or c.get("data_flow") or "",
+                "proof": v.get("proof")
+                or c.get("proof")
+                or {"input": "", "outcome": ""},
+                "status": status,
+                "recommendation": v.get("recommendation")
+                or c.get("recommendation")
+                or "",
+            }
+        )
+    return out
+
+
+def summarize(findings: list[dict]) -> dict:
+    conf_crit = sum(
+        1
+        for f in findings
+        if f["status"] == "confirmed" and f["severity"] == "critical"
+    )
+    conf_high = sum(
+        1 for f in findings if f["status"] == "confirmed" and f["severity"] == "high"
+    )
+    return {
+        "confirmed_critical": conf_crit,
+        "confirmed_high": conf_high,
+        "block": (conf_crit + conf_high) > 0,
+    }
+
+
+async def main_async(args: argparse.Namespace) -> int:
+    target = Path(args.target).resolve()
+    if not target.is_dir():
+        log(f"target dir not found: {target}")
+        return 2
+    scope = args.scope.strip() if args.scope else "the entire repository"
+    selected = (
+        [d.strip() for d in args.detectors.split(",") if d.strip()]
+        if args.detectors
+        else list(DETECTORS)
+    )
+    unknown = [d for d in selected if d not in DETECTORS]
+    if unknown:
+        log(f"unknown detector(s): {unknown}; valid: {list(DETECTORS)}")
+        return 2
+
+    budget = Budget(args.total_budget_usd)
+    sem = asyncio.Semaphore(max(1, args.concurrency))
+    log(
+        f"target={target} scope='{scope}' detectors={selected} model={args.model or 'cli-default'} "
+        f"concurrency={args.concurrency} total_budget=${args.total_budget_usd}"
+    )
+
+    det_results = await asyncio.gather(
+        *[
+            run_detector(
+                cat,
+                scope,
+                cwd=target,
+                model=args.model,
+                max_turns=args.max_turns,
+                budget_usd=args.detector_budget_usd,
+                sem=sem,
+                budget=budget,
+            )
+            for cat in selected
+        ]
+    )
+
+    candidates: list[dict] = []
+    errors: list[str] = []
+    for cat, found, err in det_results:
+        candidates.extend(found)
+        if err:
+            errors.append(f"{cat}: {err}")
+
+    log(
+        f"total candidates: {len(candidates)}; detector spend so far: ${budget.spent:.3f}"
+    )
+
+    if candidates and not budget.exhausted():
+        verdicts, _, verr = await run_verifier(
+            candidates,
+            cwd=target,
+            model=args.model,
+            max_turns=args.max_turns,
+            budget_usd=args.detector_budget_usd,
+            budget=budget,
+        )
+        if verr:
+            errors.append(f"verifier: {verr}")
+    else:
+        verdicts = {}
+        if budget.exhausted():
+            errors.append(
+                "verifier: skipped (budget exhausted) — all candidates left unverified"
+            )
+
+    findings = merge(candidates, verdicts)
+    report = {
+        "findings": findings,
+        "summary": summarize(findings),
+        "_meta": {
+            "target": str(target),
+            "scope": scope,
+            "detectors": selected,
+            "model": args.model or "cli-default",
+            "spend_usd": round(budget.spent, 4),
+            "errors": errors,
+        },
+    }
+    out = json.dumps(report, indent=2)
+    if args.out:
+        Path(args.out).write_text(out)
+        log(f"wrote {args.out}")
+    else:
+        print(out)
+
+    s = report["summary"]
+    log(
+        f"DONE: {s['confirmed_critical']} confirmed-crit, {s['confirmed_high']} confirmed-high, "
+        f"block={s['block']}, spend=${budget.spent:.3f}, errors={len(errors)}"
+    )
+    if errors:
+        for e in errors:
+            log(f"  ERROR/NOTE: {e}")
+    return 0
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(
+        description="Headless Sea Haven security detector fan-out + verifier"
+    )
+    p.add_argument("target", help="target repository directory")
+    p.add_argument(
+        "--scope", default="", help="space-separated subdirs to restrict the audit"
+    )
+    p.add_argument(
+        "--out", default="", help="write findings JSON here (default: stdout)"
+    )
+    p.add_argument("--model", default=None, help="model id (default: CLI default)")
+    p.add_argument(
+        "--detectors",
+        default="",
+        help="comma list to restrict detectors (default: all 6)",
+    )
+    p.add_argument(
+        "--concurrency", type=int, default=3, help="max concurrent detectors"
+    )
+    p.add_argument(
+        "--max-turns",
+        type=int,
+        default=40,
+        help="max agent turns per detector/verifier",
+    )
+    p.add_argument(
+        "--detector-budget-usd", type=float, default=2.0, help="per-call SDK spend cap"
+    )
+    p.add_argument(
+        "--total-budget-usd",
+        type=float,
+        default=12.0,
+        help="overall spend cap (0 = unlimited)",
+    )
+    args = p.parse_args()
+
+    # Guarantee the subscription OAuth path: a raw API key would silently win, so remove it.
+    if os.environ.pop("ANTHROPIC_API_KEY", None):
+        log("removed ANTHROPIC_API_KEY from env to force the subscription OAuth path")
+    if not os.environ.get("CLAUDE_CODE_OAUTH_TOKEN"):
+        log(
+            "FATAL: CLAUDE_CODE_OAUTH_TOKEN not set (source ~/secrev.env). Refusing to run."
+        )
+        sys.exit(2)
+
+    sys.exit(asyncio.run(main_async(args)))
+
+
+if __name__ == "__main__":
+    main()