CASAN/packages/casan-harness/scripts/bash/loop-gate.py

#!/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())