#!/usr/bin/env python3 """CASAN Per-iteration Verify Contract (Plan-17 Track 3, harness-owned). Turns the existing gates into a single per-loop-iteration contract: every iteration must pass this before the loop may advance. It composes: * H4 security (security-check.sh, input mode) — prompt-injection / secret / exfil defense; a block => DENY (terminal, never retried), * H3 verify — a deterministic success-criteria check (must_contain / must_not_contain) so "done" is *proven*, not self-declared by the model. Structured correction (Plan-17 17.10): a FAIL is retried at most `max_corrections_per_step` times (from loop-policy.yaml via the Budget Governor); exceeding that budget ESCALATES instead of retrying blindly. Fail-closed (Plan-17 17.11): a missing/unreadable artifact, a gate error, or a security-check error resolves to FAIL/DENY — never an implicit PASS. No self-declared DONE (Plan-17 17.12): `--claim-done` only yields done=true when declared success-criteria are present AND satisfied; a bare claim fails closed. Usage: loop-gate.py verify --run-id R --step N --artifact PATH \ [--success-criteria FILE] [--claim-done] \ [--profile prod|dev] [--delegation-level L2] [--project okr] Exit codes: 0 PASS (verify passed; payload.done indicates success-criteria met + claimed) 1 FAIL (verify failed; correction budget remains -> caller corrects & retries) 3 DENY (H4 security block; terminal, not retried) 4 ESCALATE (FAIL and correction budget exhausted) 2 usage error """ import argparse import json import os import subprocess import sys import tempfile import loop_common as lc def _emit(payload, stream=sys.stdout): json.dump(payload, stream, ensure_ascii=False, indent=2, sort_keys=True) stream.write("\n") def _corrections_path(run_id): return os.path.join(lc.run_dir(run_id), "corrections.json") def _load_corrections(run_id): path = _corrections_path(run_id) if os.path.isfile(path): try: return json.load(open(path, encoding="utf-8")) except ValueError: raise lc.PolicyError("corrections_state_corrupt") return {} def _bump_correction(run_id, step): path = _corrections_path(run_id) os.makedirs(os.path.dirname(path), exist_ok=True) data = _load_corrections(run_id) key = str(step) data[key] = int(data.get(key, 0)) + 1 tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(data, fh, sort_keys=True) os.replace(tmp, path) return data[key] def _run_h4(artifact_path): """Return (blocked: bool, detail: str). Fail-closed: any non-zero exit (block, error, timeout) is treated as blocked.""" gate = os.path.join(os.path.dirname(__file__), "security-check.sh") if not os.path.isfile(gate): return True, "h4_gate_missing" with tempfile.NamedTemporaryFile(prefix="loopgate-h4-", suffix=".out", delete=False) as tf: out_path = tf.name try: r = subprocess.run( ["bash", gate, artifact_path, out_path, "input"], capture_output=True, text=True, timeout=60, ) if r.returncode != 0: return True, (r.stderr.strip() or f"h4_exit_{r.returncode}")[:400] return False, "clean" except subprocess.TimeoutExpired: return True, "h4_timeout" except Exception as exc: # never fail open return True, f"h4_error:{exc}" finally: try: os.unlink(out_path) except OSError: pass def _load_criteria(path): if not path: return None if not os.path.isfile(path): raise lc.PolicyError("success_criteria_missing") try: data = json.load(open(path, encoding="utf-8")) except ValueError as exc: raise lc.PolicyError(f"success_criteria_unreadable:{exc}") if not isinstance(data, dict): raise lc.PolicyError("success_criteria_not_mapping") return data def _run_h3(text, criteria): """Deterministic faithfulness/eval: verify the artifact against declared success-criteria. Returns (passed, hint, checked).""" if not criteria: return True, None, False must = criteria.get("must_contain") or [] must_not = criteria.get("must_not_contain") or [] missing = [m for m in must if m not in text] present_bad = [m for m in must_not if m in text] if missing or present_bad: hint = {"missing": missing, "forbidden_present": present_bad} return False, hint, True return True, None, True def cmd_verify(args): prof = args.profile or lc.profile() try: policy = lc.load_policy() ceiling, _src = lc.resolve_budget(policy, prof, args.delegation_level, args.project) max_corr = int(ceiling["max_corrections_per_step"]) criteria = _load_criteria(args.success_criteria) except lc.PolicyError as exc: payload = { "verdict": "FAIL", "reason": "fail_closed", "detail": str(exc), "run_id": args.run_id, "step": args.step, "provenance": lc.provenance("loop-gate", args.artifact, verified=False), } _emit(payload) return 1 # Fail-closed: artifact must exist and be readable. if not os.path.isfile(args.artifact): payload = { "verdict": "FAIL", "reason": "artifact_unreadable", "correction_hint": "produce the artifact before verifying", "run_id": args.run_id, "step": args.step, "provenance": lc.provenance("loop-gate", args.artifact, verified=False), } _emit(payload) return 1 try: text = open(args.artifact, encoding="utf-8", errors="replace").read() except Exception as exc: payload = { "verdict": "FAIL", "reason": "artifact_read_error", "detail": str(exc), "run_id": args.run_id, "step": args.step, "provenance": lc.provenance("loop-gate", args.artifact, verified=False), } _emit(payload) return 1 # H4 security first — a security block is terminal (DENY), never retried. blocked, h4_detail = _run_h4(args.artifact) if blocked: payload = { "verdict": "DENY", "reason": "h4_security_block", "detail": h4_detail, "correction_hint": "remove injection / secret / exfil content; DENY is not retryable", "run_id": args.run_id, "step": args.step, "profile": prof, "provenance": lc.provenance("loop-gate", args.artifact, verified=policy is not None), } lc.append_audit({"kind": "gate_deny", **payload}) _emit(payload) return 3 # H3 verify against declared success-criteria. h3_pass, hint, checked = _run_h3(text, criteria) if not h3_pass: count = _bump_correction(args.run_id, args.step) if count > max_corr: payload = { "verdict": "ESCALATE", "reason": "correction_budget_exhausted", "corrections": count, "max_corrections_per_step": max_corr, "correction_hint": hint, "run_id": args.run_id, "step": args.step, "profile": prof, "provenance": lc.provenance("loop-gate", args.artifact, verified=policy is not None), } lc.append_audit({"kind": "gate_escalate", **payload}) _emit(payload) return 4 payload = { "verdict": "FAIL", "reason": "success_criteria_unmet", "corrections": count, "max_corrections_per_step": max_corr, "correction_hint": hint, "run_id": args.run_id, "step": args.step, "profile": prof, "provenance": lc.provenance("loop-gate", args.artifact, verified=policy is not None), } _emit(payload) return 1 # PASS. DONE only when success-criteria were actually checked AND the caller # claims completion — never on the model's word alone. done = bool(args.claim_done and checked) if args.claim_done and not checked: # Self-declared done without verifiable criteria => fail closed. payload = { "verdict": "FAIL", "reason": "unverifiable_done_claim", "correction_hint": "declare success-criteria (--success-criteria) to claim DONE", "run_id": args.run_id, "step": args.step, "profile": prof, "provenance": lc.provenance("loop-gate", args.artifact, verified=policy is not None), } _emit(payload) return 1 payload = { "verdict": "PASS", "done": done, "run_id": args.run_id, "step": args.step, "profile": prof, "criteria_checked": checked, "provenance": lc.provenance("loop-gate", args.artifact, verified=policy is not None), } _emit(payload) return 0 def build_parser(): p = argparse.ArgumentParser(description="CASAN Per-iteration Verify Contract (Plan-17 T3)") sub = p.add_subparsers(dest="cmd", required=True) v = sub.add_parser("verify", help="Verify one loop iteration (H4 + H3 contract)") v.add_argument("--run-id", required=True) v.add_argument("--step", type=int, required=True) v.add_argument("--artifact", required=True) v.add_argument("--success-criteria", default=None) v.add_argument("--claim-done", action="store_true") v.add_argument("--profile", default=None) v.add_argument("--delegation-level", default=None) v.add_argument("--project", default=None) v.set_defaults(func=cmd_verify) return p def main(argv=None): args = build_parser().parse_args(argv) try: return args.func(args) except lc.PolicyError as exc: _emit({"verdict": "FAIL", "reason": "fail_closed", "detail": str(exc)}) return 1 except Exception as exc: # never fail open _emit({"verdict": "FAIL", "reason": "internal_error", "detail": str(exc)}, sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())