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

#!/usr/bin/env python3
"""CASAN Meta-loop — self-improving loop policy (Plan-17 Track 5, harness-owned).

The loop that improves the loop. It reads a loop trace (and optional AgentOps
metrics) and PROPOSES loop-policy changes (dry-run, never writes). Applying a
proposal is governed exactly like every other security-sensitive change:
  * proposal != application (17.17): `propose` only prints JSON,
  * apply needs a real approval + Separation of Duties (proposer != approver) (17.18),
  * apply routes through control-plane-settings.py so the change is versioned,
    audited (hash-chain) and rollback-able,
  * a loosen proposal that exceeds the org hard cap is refused (17.19) — the
    meta-loop cannot grant itself unbounded budget.

Subcommands:
  propose --loop-trace R|FILE [--agentops JSONL] [--profile prod|dev]
  apply   --proposals FILE --id ID --proposer P --approver A [--approval TOK]
          [--allow-untrusted]

Exit codes:
  0  proposed / applied ok
  3  DENY (SoD violation, missing approval, exceeds org ceiling, untrusted, governed-set failed)
  1  unreadable proposals / unknown id
  2  usage error
"""
import argparse
import json
import os
import subprocess
import sys

import loop_common as lc


def _emit(payload):
    json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
    sys.stdout.write("\n")


def _load_trace(ref):
    """Accept either a run-id (resolved to its trace.jsonl) or a direct file path."""
    path = ref
    if not os.path.isfile(path):
        candidate = os.path.join(lc.run_dir(ref), "trace.jsonl")
        if os.path.isfile(candidate):
            path = candidate
        else:
            raise lc.PolicyError("trace_not_found")
    rows = []
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                rows.append(json.loads(line))
            except ValueError:
                raise lc.PolicyError("trace_corrupt")
    # entries are {seq, prev_hash, iteration, hash}; return the iterations
    return [r.get("iteration", r) for r in rows]


def build_proposals(iterations, agentops):
    proposals = []
    steps = [it.get("step", 0) for it in iterations if isinstance(it.get("step"), int)]
    max_step = max(steps) if steps else 0
    decisions = [str(it.get("decision") or "") for it in iterations]
    verdicts = [str(it.get("gate_verdict") or "") for it in iterations]
    progresses = [it.get("progress") for it in iterations if isinstance(it.get("progress"), (int, float))]

    # (a) The loop repeatedly hit its budget ceiling -> propose loosening max_steps.
    #     Loosening is security-sensitive and is capped by org_ceiling at apply time.
    if any(d == "HALT" for d in decisions):
        proposals.append({
            "id": "P-LOOSEN-STEPS",
            "key": "loop.max_steps",
            "value": max_step + 10,
            "direction": "loosen",
            "security_sensitive": True,
            "reason": f"run halted on budget at step {max_step}; propose raising max_steps to {max_step + 10}",
        })

    # (b) Oscillation/thrash seen -> propose TIGHTENING oscillation_repeat (detect sooner).
    #     Tightening is safe but still governed (propose != apply).
    if any(v == "OSCILLATING" for v in verdicts) or (agentops or {}).get("oscillations"):
        proposals.append({
            "id": "P-TIGHTEN-OSC",
            "key": "loop.oscillation_repeat",
            "value": 2,
            "direction": "tighten",
            "security_sensitive": False,
            "reason": "oscillation observed; tighten oscillation_repeat to 2 to break loops sooner",
        })

    # (c) Progress stagnated across the trace -> propose tightening the no-progress window.
    if len(progresses) >= 3 and max(progresses) <= min(progresses):
        proposals.append({
            "id": "P-TIGHTEN-NOPROG",
            "key": "loop.no_progress_window",
            "value": 2,
            "direction": "tighten",
            "security_sensitive": False,
            "reason": "no forward progress across the trace; tighten no_progress_window to 2",
        })
    return proposals


def cmd_propose(args):
    iterations = _load_trace(args.loop_trace)
    agentops = None
    if args.agentops and os.path.isfile(args.agentops):
        try:
            agentops = json.load(open(args.agentops, encoding="utf-8"))
        except ValueError:
            agentops = None
    # AgentOps/loop-trace here is local, unsigned telemetry -> untrusted by default
    # (ARCH-08 parity with self-improve). Enforced apply of an untrusted proposal
    # requires --allow-untrusted.
    proposals = build_proposals(iterations, agentops)
    for p in proposals:
        p["source_trust"] = "untrusted"
    _emit({"proposals": proposals, "count": len(proposals), "source_trust": "untrusted"})
    return 0


def cmd_apply(args):
    try:
        data = json.load(open(args.proposals, encoding="utf-8"))
    except (OSError, ValueError):
        _emit({"decision": "DENY", "reason": "proposals_unreadable", "proposals": args.proposals})
        return 1
    proposal = next((p for p in data.get("proposals", []) if p.get("id") == args.id), None)
    if proposal is None:
        _emit({"decision": "DENY", "reason": "unknown_proposal", "id": args.id})
        return 1

    # Separation of Duties (17.18): the proposer can never be their own approver.
    if not args.proposer or not args.approver or args.proposer == args.approver:
        _emit({"decision": "DENY", "reason": "sod_violation",
               "proposer": args.proposer, "approver": args.approver})
        return 3

    # Applying ALWAYS requires an approval (propose != apply).
    if not (args.approval or "").strip():
        _emit({"decision": "DENY", "reason": "approval_required", "id": args.id})
        return 3

    # ARCH-08: refuse untrusted telemetry-derived proposals in enforced mode.
    enforced = (os.environ.get("CASAN_PROFILE") == "prod"
                or os.environ.get("CASAN_METALOOP_STRICT") == "1")
    if proposal.get("source_trust", "untrusted") == "untrusted" and enforced and not args.allow_untrusted:
        _emit({"decision": "DENY", "reason": "untrusted_source", "id": args.id})
        return 3

    key = proposal.get("key")
    value = proposal.get("value")
    if key not in lc._GOV_BUDGET_MAP and key not in lc._GOV_CONV_MAP:
        _emit({"decision": "DENY", "reason": "key_not_governable", "key": key})
        return 3

    # Org hard cap (17.19): a loosen can never exceed the org ceiling.
    if proposal.get("direction") == "loosen" and key in lc._GOV_BUDGET_MAP:
        try:
            policy = lc.load_policy()
        except lc.PolicyError as exc:
            _emit({"decision": "DENY", "reason": "fail_closed", "detail": str(exc)})
            return 3
        oc = lc.org_ceiling(policy)
        bk = lc._GOV_BUDGET_MAP[key]
        if bk in oc and isinstance(value, (int, float)) and value > oc[bk]:
            _emit({"decision": "DENY", "reason": "exceeds_org_ceiling",
                   "key": key, "value": value, "org_ceiling": oc[bk]})
            return 3

    # Governed apply: route through the control-plane store (versioned + audit +
    # rollback + its own approval verification in enforced mode). CASAN_APPROVER is
    # honoured by control-plane-settings.py's enforced approval check.
    cps = os.path.join(os.path.dirname(__file__), "control-plane-settings.py")
    env = dict(os.environ)
    env["CASAN_APPROVER"] = args.approver
    cmd = [
        sys.executable, cps, "set", key, json.dumps(value),
        "--actor", args.proposer,
        "--reason", f"meta-loop {args.id} ({proposal.get('direction')})",
        "--approval", args.approval,
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, env=env)
    if result.returncode != 0:
        _emit({"decision": "DENY", "reason": "governed_set_failed",
               "detail": result.stderr.strip(), "id": args.id})
        return 3
    lc.append_audit({"kind": "metaloop_applied", "id": args.id, "key": key,
                     "value": value, "proposer": args.proposer, "approver": args.approver})
    _emit({"decision": "APPLIED", "id": args.id, "key": key, "value": value,
           "proposer": args.proposer, "approver": args.approver})
    return 0


def build_parser():
    p = argparse.ArgumentParser(description="CASAN Meta-loop (Plan-17 T5)")
    sub = p.add_subparsers(dest="cmd", required=True)

    pr = sub.add_parser("propose", help="Propose loop-policy changes (dry-run)")
    pr.add_argument("--loop-trace", required=True, help="run-id or trace.jsonl path")
    pr.add_argument("--agentops", default=None)
    pr.add_argument("--profile", default=None)
    pr.set_defaults(func=cmd_propose)

    ap = sub.add_parser("apply", help="Apply an approved proposal (governed + SoD)")
    ap.add_argument("--proposals", required=True)
    ap.add_argument("--id", required=True)
    ap.add_argument("--proposer", required=True)
    ap.add_argument("--approver", required=True)
    ap.add_argument("--approval", default="")
    ap.add_argument("--allow-untrusted", action="store_true")
    ap.set_defaults(func=cmd_apply)
    return p


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


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