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

#!/usr/bin/env python3
"""CASAN Loop Trace / Replay (Plan-17 Track 4, harness-owned).

Records each loop iteration (§3 Loop Contract) into a per-run, append-only,
hash-linked trace so a whole loop can be audited and deterministically replayed.
Gives loops the "loop view" the audit chain lacked.

Subcommands:
  record       append one Iteration to the run's trace (hash-linked to the prev).
  show         render the loop view (JSON on stdout; a human table on stderr).
  replay       re-run the per-iteration Verify Contract (loop-gate) on each
               recorded artifact and compare the fresh verdict to the recorded
               one -> detects non-determinism / tampered artifacts.
  verify-chain recompute the hash chain -> any edited/removed record BREAKs it.

Fail-closed & tail-safe (Plan-17 §2): the trace is append-only; replay never
mutates the source trace (it runs loop-gate under throwaway run-ids). A corrupt
trace file fails closed (verify-chain BREAK / show error).

KMS-anchoring of the chain head (17.16 full) is a TIER-2 add-on (Plan-07 B3 / A7);
this offline slice proves local tamper-evidence without it.

Usage:
  loop-trace.py record --run-id R --step N --intent ... --action ... \
      [--tool T] [--inputs-ref REF] [--gate-verdict PASS|FAIL|DENY|ESCALATE] \
      [--progress P] [--decision CONTINUE|HALT|ESCALATE|DONE] \
      [--artifact PATH] [--success-criteria FILE] [--evidence-ref REF] \
      [--budget-snapshot JSON]
  loop-trace.py show        --run-id R [--json]
  loop-trace.py replay      --run-id R [--profile prod|dev]
  loop-trace.py verify-chain --run-id R

Exit codes:
  0  ok / chain intact / replay all-match
  3  chain BREAK, replay drift detected, or fail-closed error
  2  usage error
"""
import argparse
import json
import os
import subprocess
import sys

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 _trace_path(run_id):
    return os.path.join(lc.run_dir(run_id), "trace.jsonl")


def _read_entries(run_id):
    """Return the list of raw chain entries (each {seq, prev_hash, iteration, hash}).
    A malformed line fails closed."""
    path = _trace_path(run_id)
    entries = []
    if os.path.isfile(path):
        with open(path, encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    entries.append(json.loads(line))
                except ValueError:
                    raise lc.PolicyError("trace_corrupt")
    return entries


def cmd_record(args):
    budget = None
    if args.budget_snapshot:
        try:
            budget = json.loads(args.budget_snapshot)
        except ValueError:
            raise lc.PolicyError("bad_budget_snapshot_json")

    iteration = {
        "run_id": args.run_id,
        "step": args.step,
        "intent": args.intent,
        "action": args.action,
        "tool": args.tool,
        "inputs_ref": args.inputs_ref,
        "gate_verdict": args.gate_verdict,
        "progress": args.progress,
        "budget_snapshot": budget,
        "decision": args.decision,
        "artifact": args.artifact,
        "success_criteria": args.success_criteria,
        "evidence_ref": args.evidence_ref,
        "ts": lc.now_iso(),
    }

    path = _trace_path(args.run_id)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    entries = _read_entries(args.run_id)
    prev = entries[-1]["hash"] if entries else lc.GENESIS_HASH
    seq = len(entries)
    base = {"seq": seq, "prev_hash": prev, "iteration": iteration}
    base["hash"] = lc.chain_hash(base)
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(lc.canonical(base) + "\n")
    _emit({"decision": "RECORDED", "run_id": args.run_id, "step": args.step, "seq": seq})
    return 0


def cmd_show(args):
    entries = _read_entries(args.run_id)
    iterations = [e["iteration"] for e in entries]
    payload = {
        "run_id": args.run_id,
        "count": len(iterations),
        "iterations": iterations,
        "provenance": lc.provenance("loop-trace", _trace_path(args.run_id), verified=False),
    }
    # Human-readable loop view on stderr; machine JSON on stdout.
    hdr = f"{'seq':>3} {'step':>4} {'verdict':<9} {'decision':<10} {'progress':>8}  intent"
    print(hdr, file=sys.stderr)
    for e in entries:
        it = e["iteration"]
        print(
            f"{e['seq']:>3} {str(it.get('step','')):>4} "
            f"{str(it.get('gate_verdict','')):<9} {str(it.get('decision','')):<10} "
            f"{str(it.get('progress','')):>8}  {str(it.get('intent',''))[:48]}",
            file=sys.stderr,
        )
    _emit(payload)
    return 0


def cmd_verify_chain(args):
    try:
        entries = _read_entries(args.run_id)
    except lc.PolicyError as exc:
        _emit({"decision": "BREAK", "reason": "fail_closed", "detail": str(exc), "run_id": args.run_id})
        return 3
    prev = lc.GENESIS_HASH
    for i, e in enumerate(entries):
        stored = e.get("hash")
        base = {"seq": e.get("seq"), "prev_hash": e.get("prev_hash"), "iteration": e.get("iteration")}
        recomputed = lc.chain_hash(base)
        if e.get("prev_hash") != prev or e.get("seq") != i or recomputed != stored:
            _emit({
                "decision": "BREAK", "reason": "chain_broken", "at_seq": i,
                "run_id": args.run_id,
                "provenance": lc.provenance("loop-trace", _trace_path(args.run_id), verified=False),
            })
            return 3
        prev = stored
    _emit({
        "decision": "OK", "run_id": args.run_id, "records": len(entries),
        "provenance": lc.provenance("loop-trace", _trace_path(args.run_id), verified=True),
    })
    return 0


def cmd_replay(args):
    try:
        entries = _read_entries(args.run_id)
    except lc.PolicyError as exc:
        _emit({"decision": "FAIL", "reason": "fail_closed", "detail": str(exc), "run_id": args.run_id})
        return 3

    gate = os.path.join(os.path.dirname(__file__), "loop-gate.py")
    diffs = []
    replayed = 0
    for e in entries:
        it = e["iteration"]
        artifact = it.get("artifact")
        recorded = it.get("gate_verdict")
        if not artifact or not recorded:
            continue  # only iterations with a verifiable artifact + verdict
        replayed += 1
        cmd = [
            sys.executable, gate, "verify",
            "--run-id", f"{args.run_id}-replay-{e['seq']}",
            "--step", str(it.get("step", 0)),
            "--artifact", artifact,
        ]
        if it.get("success_criteria"):
            cmd += ["--success-criteria", it["success_criteria"]]
        if args.profile:
            cmd += ["--profile", args.profile]
        try:
            r = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
            fresh = json.loads(r.stdout).get("verdict") if r.stdout.strip() else "ERROR"
        except Exception as exc:
            fresh = f"ERROR:{exc}"
        if fresh != recorded:
            diffs.append({"seq": e["seq"], "step": it.get("step"), "recorded": recorded, "replayed": fresh})

    payload = {
        "run_id": args.run_id,
        "replayed": replayed,
        "diffs": diffs,
        "provenance": lc.provenance("loop-trace", _trace_path(args.run_id), verified=not diffs),
    }
    if diffs:
        payload["decision"] = "DRIFT"
        payload["reason"] = "verdict_mismatch"
        lc.append_audit({"kind": "trace_replay_drift", **payload})
        _emit(payload)
        return 3
    payload["decision"] = "MATCH"
    _emit(payload)
    return 0


def build_parser():
    p = argparse.ArgumentParser(description="CASAN Loop Trace / Replay (Plan-17 T4)")
    sub = p.add_subparsers(dest="cmd", required=True)

    r = sub.add_parser("record", help="Append one Iteration to the run trace")
    r.add_argument("--run-id", required=True)
    r.add_argument("--step", type=int, required=True)
    r.add_argument("--intent", default=None)
    r.add_argument("--action", default=None)
    r.add_argument("--tool", default=None)
    r.add_argument("--inputs-ref", default=None)
    r.add_argument("--gate-verdict", default=None)
    r.add_argument("--progress", type=float, default=None)
    r.add_argument("--decision", default=None)
    r.add_argument("--artifact", default=None)
    r.add_argument("--success-criteria", default=None)
    r.add_argument("--evidence-ref", default=None)
    r.add_argument("--budget-snapshot", default=None)
    r.set_defaults(func=cmd_record)

    s = sub.add_parser("show", help="Render the loop view")
    s.add_argument("--run-id", required=True)
    s.add_argument("--json", action="store_true")
    s.set_defaults(func=cmd_show)

    rp = sub.add_parser("replay", help="Deterministically re-verify recorded artifacts")
    rp.add_argument("--run-id", required=True)
    rp.add_argument("--profile", default=None)
    rp.set_defaults(func=cmd_replay)

    vc = sub.add_parser("verify-chain", help="Recompute the hash chain (tamper-evidence)")
    vc.add_argument("--run-id", required=True)
    vc.set_defaults(func=cmd_verify_chain)
    return p


def main(argv=None):
    args = build_parser().parse_args(argv)
    try:
        return args.func(args)
    except lc.PolicyError as exc:
        _emit({"decision": "FAIL", "reason": "fail_closed", "detail": str(exc)})
        return 3
    except Exception as exc:  # never fail open
        _emit({"decision": "FAIL", "reason": "internal_error", "detail": str(exc)}, sys.stderr)
        return 3


if __name__ == "__main__":
    raise SystemExit(main())