gitea-admin/CASAN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update plan 17

Browse Source
This commit is contained in:
thanhnv
2026-07-07 18:58:21 +09:00
parent ae4fc7112c
commit 5fbc017a2c
23 changed files with 2649 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AINative_OKR_CASAN5/.specify/scripts/bash/ci-harness-gate.sh
+8
View File
@@ -126,6 +126,14 @@ run "phase-sec23-registry-crypt" bash "$TESTS/phase-sec23-registry-crypt-tests.s
run "phase-sec24-supplychain" bash "$TESTS/phase-sec24-tests.sh"
run "phase-sec25-attestation" bash "$TESTS/phase-sec25-tests.sh"
# Plan-17 loop engineering (Agentic Loop Governance) — each primitive fail-closed.
run "phase-loop-governor" bash "$TESTS/phase-loop-governor-tests.sh"
run "phase-loop-convergence" bash "$TESTS/phase-loop-convergence-tests.sh"
run "phase-loop-gate" bash "$TESTS/phase-loop-gate-tests.sh"
run "phase-loop-trace" bash "$TESTS/phase-loop-trace-tests.sh"
run "phase-loop-metaloop" bash "$TESTS/phase-loop-metaloop-tests.sh"
run "phase-loop-run" bash "$TESTS/phase-loop-run-tests.sh"
# ARCH-02: coverage cannot silently drop; ARCH-01: harness/policy cannot silently
# drift. Both SKIP cleanly when no manifest is provisioned (non-strict dev/CI).
run "test-integrity" python3 "$SCRIPT_DIR/test-integrity.py" verify
AINative_OKR_CASAN5/.specify/scripts/bash/control-plane-settings.py
+10
View File
@@ -35,6 +35,16 @@ SETTINGS_POLICY = {
"model.primary": {"securitySensitive": False, "description": "Primary model spec (e.g. ollama:ornith:9b)"},
"security.strict": {"securitySensitive": True, "description": "H4 fail-closed strict mode"},
"kill_switch.global": {"securitySensitive": True, "description": "Global kill-switch engage/disengage"},
# Plan-17 loop governance overrides (meta-loop, T5). Loosening a loop budget /
# widening a convergence window is security-sensitive: it grants the agent more
# autonomy, so it needs a real approval + SoD and is clamped to org_ceiling.
"loop.max_steps": {"securitySensitive": True, "description": "Loop Governor: max steps per run"},
"loop.max_tokens": {"securitySensitive": True, "description": "Loop Governor: max tokens per run"},
"loop.max_wall_clock_sec": {"securitySensitive": True, "description": "Loop Governor: max wall-clock seconds per run"},
"loop.max_cost_usd": {"securitySensitive": True, "description": "Loop Governor: max cost USD per run"},
"loop.max_corrections_per_step": {"securitySensitive": True, "description": "Loop Governor: max corrections per step"},
"loop.oscillation_repeat": {"securitySensitive": True, "description": "Convergence: identical-action repeats before OSCILLATING"},
"loop.no_progress_window": {"securitySensitive": True, "description": "Convergence: no-progress window before STALLED"},
}
GENESIS_HASH = "0" * 64
AINative_OKR_CASAN5/.specify/scripts/bash/loop-convergence.py
+206
View File
@@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""CASAN No-progress / Oscillation Detector (Plan-17 Track 2, harness-owned).
Watches a run's per-step observations and decides whether the loop is CONVERGING,
STALLED (no forward progress), or OSCILLATING (repeating/thrashing actions). A
stalled or oscillating loop is halted or escalated (on_stall policy) instead of
being allowed to burn budget going nowhere.
Detection rules (thresholds from loop-policy.yaml, else strictest built-in):
* oscillation = the last N observations share one action_hash (stuck repeating),
* thrash = the last `thrash_window` steps alternate between exactly 2
action_hashes (A,B,A,B...),
* no-progress = the last W observations made zero forward progress
(progress never exceeded the running best).
Deny-by-default & fail-closed: absent/unknown policy => strictest thresholds
(detect sooner); a corrupt policy => ESCALATE/HALT, never silent CONVERGING.
Insufficient data (< window) => CONVERGING (the Budget Governor is the hard stop).
Usage:
loop-convergence.py observe --run-id R --step N --action-hash H --progress P
loop-convergence.py verdict --run-id R [--profile prod|dev]
Exit codes:
0 CONVERGING (or observe recorded ok)
3 HALT (STALLED/OSCILLATING with on_stall=halt, OR fail-closed error)
4 ESCALATE (STALLED/OSCILLATING with on_stall=escalate)
2 usage error
"""
import argparse
import json
import os
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 _obs_path(run_id):
return os.path.join(lc.run_dir(run_id), "convergence.jsonl")
def _load_obs(run_id):
path = _obs_path(run_id)
rows = []
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:
rows.append(json.loads(line))
except ValueError:
# A corrupt observation stream cannot be trusted.
raise lc.PolicyError("observation_stream_corrupt")
rows.sort(key=lambda r: r.get("step", 0))
return rows
def cmd_observe(args):
path = _obs_path(args.run_id)
os.makedirs(os.path.dirname(path), exist_ok=True)
rec = {
"step": args.step,
"action_hash": args.action_hash,
"progress": args.progress,
"ts": lc.now_iso(),
}
with open(path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(rec, ensure_ascii=False, sort_keys=True) + "\n")
_emit({"decision": "OBSERVED", "run_id": args.run_id, "step": args.step})
return 0
def _detect_oscillation(hashes, cfg):
n = cfg["oscillation_repeat"]
if len(hashes) >= n and len(set(hashes[-n:])) == 1:
return {"pattern": "repeat", "action_hash": hashes[-1], "count": n}
w = cfg["thrash_window"]
if len(hashes) >= w:
window = hashes[-w:]
distinct = set(window)
if len(distinct) == 2:
# A,B,A,B... : every element differs from its immediate neighbour.
if all(window[i] != window[i + 1] for i in range(len(window) - 1)):
return {"pattern": "thrash", "actions": sorted(distinct), "window": w}
return None
def _detect_no_progress(progresses, cfg):
w = cfg["no_progress_window"]
if len(progresses) < w:
return None
# Mark each step that improved on the running best; STALLED if the last W
# steps contain no improvement at all.
best = None
improved = []
for p in progresses:
if p is None:
improved.append(False)
continue
if best is None or p > best:
improved.append(True)
best = p
else:
improved.append(False)
if not any(improved[-w:]):
return {"window": w, "last_progress": progresses[-1]}
return None
def cmd_verdict(args):
prof = args.profile or lc.profile()
try:
policy = lc.load_policy()
cfg, source = lc.resolve_convergence(policy, prof)
rows = _load_obs(args.run_id)
except lc.PolicyError as exc:
payload = {
"decision": "HALT",
"reason": "fail_closed",
"detail": str(exc),
"run_id": args.run_id,
"provenance": lc.provenance("loop-convergence", None, verified=False),
}
lc.append_audit({"kind": "convergence_halt", **payload})
_emit(payload)
return 3
hashes = [r.get("action_hash") for r in rows]
progresses = [r.get("progress") for r in rows]
osc = _detect_oscillation(hashes, cfg)
stall = None if osc else _detect_no_progress(progresses, cfg)
if osc or stall:
verdict = "OSCILLATING" if osc else "STALLED"
on_stall = cfg.get("on_stall", "escalate")
decision = "HALT" if on_stall == "halt" else "ESCALATE"
payload = {
"decision": decision,
"verdict": verdict,
"run_id": args.run_id,
"profile": prof,
"policy_source": source,
"thresholds": cfg,
"observations": len(rows),
"evidence": osc or stall,
"on_stall": on_stall,
"provenance": lc.provenance("loop-convergence", None, verified=policy is not None),
}
lc.append_audit({"kind": "convergence_" + verdict.lower(), **payload})
_emit(payload)
return 3 if decision == "HALT" else 4
payload = {
"decision": "CONVERGING",
"verdict": "CONVERGING",
"run_id": args.run_id,
"profile": prof,
"policy_source": source,
"thresholds": cfg,
"observations": len(rows),
"provenance": lc.provenance("loop-convergence", None, verified=policy is not None),
}
_emit(payload)
return 0
def build_parser():
p = argparse.ArgumentParser(description="CASAN Convergence Detector (Plan-17 T2)")
sub = p.add_subparsers(dest="cmd", required=True)
o = sub.add_parser("observe", help="Record a per-step observation")
o.add_argument("--run-id", required=True)
o.add_argument("--step", type=int, required=True)
o.add_argument("--action-hash", required=True)
o.add_argument("--progress", type=float, required=True)
o.set_defaults(func=cmd_observe)
v = sub.add_parser("verdict", help="Classify the run: CONVERGING|STALLED|OSCILLATING")
v.add_argument("--run-id", required=True)
v.add_argument("--profile", default=None)
v.set_defaults(func=cmd_verdict)
return p
def main(argv=None):
args = build_parser().parse_args(argv)
try:
return args.func(args)
except lc.PolicyError as exc:
_emit({"decision": "HALT", "reason": "fail_closed", "detail": str(exc)})
return 3
except Exception as exc: # never fail open
_emit({"decision": "HALT", "reason": "internal_error", "detail": str(exc)}, sys.stderr)
return 3
if __name__ == "__main__":
raise SystemExit(main())
AINative_OKR_CASAN5/.specify/scripts/bash/loop-gate.py
+262
View File
@@ -0,0 +1,262 @@
#!/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())
AINative_OKR_CASAN5/.specify/scripts/bash/loop-governor.py
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""CASAN Loop Budget Governor (Plan-17 Track 1, harness-owned).
The loop-breaker. Given a run's *cumulative* usage (steps / tokens / wall-clock /
cost / corrections), decide whether the agent loop may CONTINUE or must HALT.
Deny-by-default & fail-closed:
* no policy file -> strictest built-in ceiling (STRICT_CEILING),
* present-but-corrupt policy -> HALT (exit 3), never "run on",
* any unexpected error -> HALT (exit 3).
Loosening the ceiling is a security-sensitive setting: it must be changed through
the governed settings store (control-plane-settings.py: approval JWT + SoD +
versioned + rollback), not by editing this script. This governor only *reads* the
(governed) policy — see Plan-17 17.3.
Usage:
loop-governor.py check --run-id R --step N \
[--tokens T] [--elapsed S] [--cost C] [--corrections K] \
[--profile prod|dev] [--delegation-level L2] [--project okr]
Exit codes:
0 CONTINUE (within ceiling)
3 HALT(budget) (ceiling exceeded with on_exceed=halt, OR fail-closed error)
4 ESCALATE (ceiling exceeded with on_exceed=escalate)
2 usage error
"""
import argparse
import json
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 cmd_check(args):
usage = {
"steps": args.step,
"tokens": args.tokens,
"elapsed_s": args.elapsed,
"cost_usd": args.cost,
"corrections": args.corrections,
}
prof = args.profile or lc.profile()
try:
policy = lc.load_policy()
ceiling, source = lc.resolve_budget(
policy, prof, args.delegation_level, args.project
)
except lc.PolicyError as exc:
# Fail-closed: a policy we cannot trust must stop the loop, not free it.
payload = {
"decision": "HALT",
"reason": "fail_closed",
"detail": str(exc),
"run_id": args.run_id,
"step": args.step,
"profile": prof,
"provenance": lc.provenance("loop-governor", lc.policy_path(), verified=False),
}
lc.append_audit({"kind": "governor_halt", **payload})
_emit(payload)
return 3
exceeded = []
checks = (
("steps", "max_steps", usage["steps"]),
("tokens", "max_tokens", usage["tokens"]),
("elapsed_s", "max_wall_clock_sec", usage["elapsed_s"]),
("cost_usd", "max_cost_usd", usage["cost_usd"]),
("corrections", "max_corrections_per_step", usage["corrections"]),
)
for usage_key, limit_key, value in checks:
limit = ceiling[limit_key]
if value is not None and value > limit:
exceeded.append({"metric": usage_key, "value": value, "limit": limit})
on_exceed = ceiling.get("on_exceed", "halt")
if not exceeded:
payload = {
"decision": "CONTINUE",
"run_id": args.run_id,
"step": args.step,
"profile": prof,
"policy_source": source,
"limits": ceiling,
"usage": usage,
"provenance": lc.provenance("loop-governor", lc.policy_path(), verified=policy is not None),
}
_emit(payload)
return 0
decision = "ESCALATE" if on_exceed == "escalate" else "HALT"
payload = {
"decision": decision,
"reason": "budget_exceeded",
"run_id": args.run_id,
"step": args.step,
"profile": prof,
"policy_source": source,
"limits": ceiling,
"usage": usage,
"exceeded": exceeded,
"on_exceed": on_exceed,
"provenance": lc.provenance("loop-governor", lc.policy_path(), verified=policy is not None),
}
lc.append_audit({"kind": "governor_" + decision.lower(), **payload})
_emit(payload)
return 4 if decision == "ESCALATE" else 3
def build_parser():
p = argparse.ArgumentParser(description="CASAN Loop Budget Governor (Plan-17 T1)")
sub = p.add_subparsers(dest="cmd", required=True)
c = sub.add_parser("check", help="Evaluate cumulative usage against the ceiling")
c.add_argument("--run-id", required=True)
c.add_argument("--step", type=int, required=True)
c.add_argument("--tokens", type=int, default=None)
c.add_argument("--elapsed", type=float, default=None)
c.add_argument("--cost", type=float, default=None)
c.add_argument("--corrections", type=int, default=None)
c.add_argument("--profile", default=None)
c.add_argument("--delegation-level", default=None)
c.add_argument("--project", default=None)
c.set_defaults(func=cmd_check)
return p
def main(argv=None):
args = build_parser().parse_args(argv)
try:
return args.func(args)
except lc.PolicyError as exc:
_emit({"decision": "HALT", "reason": "fail_closed", "detail": str(exc)})
return 3
except Exception as exc: # never fail open
_emit({"decision": "HALT", "reason": "internal_error", "detail": str(exc)}, sys.stderr)
return 3
if __name__ == "__main__":
raise SystemExit(main())
AINative_OKR_CASAN5/.specify/scripts/bash/loop-metaloop.py
+233
View File
@@ -0,0 +1,233 @@
#!/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())
AINative_OKR_CASAN5/.specify/scripts/bash/loop-run.sh
+155
View File
@@ -0,0 +1,155 @@
#!/usr/bin/env bash
set -uo pipefail
# CASAN Loop Orchestrator (Plan-17 Track 6, harness-owned).
#
# Drives one governed agent loop. Every turn runs the loop primitives in order
# (17.20): loop-gate (verify) -> loop-governor (budget) -> loop-convergence
# (progress) -> loop-trace (record). The loop stops on the FIRST stop-condition:
# DONE gate PASS on the artifact (success-criteria proven),
# HALT governor budget exceeded (loop-breaker) or gate DENY (security),
# ESCALATE convergence STALLED/OSCILLATING (or gate ESCALATE).
#
# Secure-by-default (17.20): governance is ON. In profile=prod, disabling it
# (CASAN_LOOP_GOVERNANCE=off) is refused unless an explicit, audited opt-out
# reason is given (CASAN_LOOP_OPTOUT_REASON) — reversing the ARCH-03 lesson.
#
# Between-turn context compaction (17.21) uses context-compress.py + must-keep
# when --context is supplied, so the loop's growing context is kept bounded.
#
# Usage:
# loop-run.sh --run-id R --artifact FILE [--success-criteria FILE] \
# [--profile prod|dev] [--delegation-level L2] [--project okr] \
# [--max-steps N] [--tokens-per-step T] [--cost-per-step C] \
# [--context FILE]
#
# Exit codes: 0 DONE · 3 HALT/ESCALATE/DENY (governance stopped the loop) ·
# 4 opt-out refused (prod) · 2 usage error.
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
GATE="$SCRIPT_DIR/loop-gate.py"
GOV="$SCRIPT_DIR/loop-governor.py"
CONV="$SCRIPT_DIR/loop-convergence.py"
TRACE="$SCRIPT_DIR/loop-trace.py"
COMPRESS="$SCRIPT_DIR/context-compress.py"
RUN_ID=""; ARTIFACT=""; CRIT=""; PROFILE=""; DLEVEL=""; PROJECT=""
MAX_STEPS=25; TOKENS_PER_STEP=1000; COST_PER_STEP="0.01"; CONTEXT=""
while [[ "$#" -gt 0 ]]; do
case "$1" in
--run-id) RUN_ID="${2:-}"; shift 2 ;;
--artifact) ARTIFACT="${2:-}"; shift 2 ;;
--success-criteria) CRIT="${2:-}"; shift 2 ;;
--profile) PROFILE="${2:-}"; shift 2 ;;
--delegation-level) DLEVEL="${2:-}"; shift 2 ;;
--project) PROJECT="${2:-}"; shift 2 ;;
--max-steps) MAX_STEPS="${2:-}"; shift 2 ;;
--tokens-per-step) TOKENS_PER_STEP="${2:-}"; shift 2 ;;
--cost-per-step) COST_PER_STEP="${2:-}"; shift 2 ;;
--context) CONTEXT="${2:-}"; shift 2 ;;
*) echo "loop-run: unknown arg $1" >&2; exit 2 ;;
esac
done
if [[ -z "$RUN_ID" || -z "$ARTIFACT" ]]; then
echo "Usage: loop-run.sh --run-id R --artifact FILE [--success-criteria FILE] ..." >&2
exit 2
fi
PROFILE="${PROFILE:-${CASAN_PROFILE:-dev}}"
prof_args=(--profile "$PROFILE")
[[ -n "$DLEVEL" ]] && prof_args+=(--delegation-level "$DLEVEL")
[[ -n "$PROJECT" ]] && prof_args+=(--project "$PROJECT")
sha_of() {
if command -v sha256sum >/dev/null 2>&1; then sha256sum "$1" 2>/dev/null | awk '{print $1}'
else shasum -a 256 "$1" 2>/dev/null | awk '{print $1}'; fi
}
json_field() { sed -n "s/.*\"$1\": \"\\([A-Za-z_]*\\)\".*/\\1/p" | head -1; }
# --- secure-by-default governance gate (17.20) ------------------------------
GOVERNANCE="${CASAN_LOOP_GOVERNANCE:-on}"
if [[ "$GOVERNANCE" == "off" ]]; then
if [[ "$PROFILE" == "prod" && -z "${CASAN_LOOP_OPTOUT_REASON:-}" ]]; then
echo "LOOP_REFUSE opt-out of loop governance requires CASAN_LOOP_OPTOUT_REASON in prod" >&2
exit 4
fi
# An allowed opt-out is always audited (never silent).
PYTHONPATH="$SCRIPT_DIR" python3 - "$RUN_ID" "${CASAN_LOOP_OPTOUT_REASON:-dev-optout}" <<'PY'
import sys, loop_common as lc
lc.append_audit({"kind": "loop_governance_optout", "run_id": sys.argv[1], "reason": sys.argv[2]})
PY
fi
echo "===== loop-run run_id=$RUN_ID profile=$PROFILE governance=$GOVERNANCE ====="
FINAL="DONE"; RC=0
CUM_TOKENS=0
COST_ACC="0"
for (( step=1; step<=MAX_STEPS; step++ )); do
CUM_TOKENS=$(( CUM_TOKENS + TOKENS_PER_STEP ))
COST_ACC="$(python3 -c "print(round($COST_ACC + $COST_PER_STEP, 6))")"
DECISION="CONTINUE"; VERDICT=""; PROGRESS="0"
if [[ "$GOVERNANCE" == "on" ]]; then
# 1) Governor: cumulative budget check (loop-breaker) BEFORE more work.
set +e
python3 "$GOV" check --run-id "$RUN_ID" --step "$step" \
--tokens "$CUM_TOKENS" --cost "$COST_ACC" "${prof_args[@]}" >/dev/null 2>&1
grc=$?
set -e 2>/dev/null || true
if [[ "$grc" -eq 3 ]]; then FINAL="HALT"; DECISION="HALT"; RC=3; fi
if [[ "$grc" -eq 4 ]]; then FINAL="ESCALATE"; DECISION="ESCALATE"; RC=3; fi
if [[ "$DECISION" == "CONTINUE" ]]; then
# 2) Gate: per-iteration verify contract.
set +e
GATE_OUT="$(python3 "$GATE" verify --run-id "$RUN_ID" --step "$step" \
--artifact "$ARTIFACT" ${CRIT:+--success-criteria "$CRIT"} "${prof_args[@]}" 2>/dev/null)"
set -e 2>/dev/null || true
VERDICT="$(printf '%s' "$GATE_OUT" | json_field verdict)"
case "$VERDICT" in
PASS) FINAL="DONE"; DECISION="DONE"; PROGRESS="1"; RC=0 ;;
DENY) FINAL="HALT"; DECISION="HALT"; RC=3 ;;
ESCALATE) FINAL="ESCALATE"; DECISION="ESCALATE"; RC=3 ;;
*) DECISION="CONTINUE"; PROGRESS="0" ;; # FAIL -> keep correcting
esac
fi
# 3) Convergence: observe + verdict (no-progress / oscillation breaker).
if [[ "$DECISION" == "CONTINUE" ]]; then
AH="$(sha_of "$ARTIFACT")"; AH="${AH:0:16}"
python3 "$CONV" observe --run-id "$RUN_ID" --step "$step" \
--action-hash "$AH" --progress "$PROGRESS" >/dev/null 2>&1
set +e
python3 "$CONV" verdict --run-id "$RUN_ID" "${prof_args[@]}" >/dev/null 2>&1
cvrc=$?
set -e 2>/dev/null || true
if [[ "$cvrc" -eq 3 ]]; then FINAL="HALT"; DECISION="HALT"; RC=3; fi
if [[ "$cvrc" -eq 4 ]]; then FINAL="ESCALATE"; DECISION="ESCALATE"; RC=3; fi
fi
else
# Governance opted out: record turns only, terminate at max-steps as DONE.
DECISION="CONTINUE"
fi
# 4) Trace: append the immutable iteration record.
python3 "$TRACE" record --run-id "$RUN_ID" --step "$step" \
--intent "turn-$step" --action verify --tool loop-run \
${VERDICT:+--gate-verdict "$VERDICT"} --decision "$DECISION" --progress "$PROGRESS" \
--artifact "$ARTIFACT" ${CRIT:+--success-criteria "$CRIT"} \
--budget-snapshot "{\"steps\":$step,\"tokens\":$CUM_TOKENS,\"cost_usd\":$COST_ACC}" >/dev/null 2>&1
# 5) Between-turn context compaction (17.21).
if [[ -n "$CONTEXT" && -f "$CONTEXT" ]]; then
python3 "$COMPRESS" --mode structural --input "$CONTEXT" \
> "$(dirname "$ARTIFACT")/.loop-context-compacted.txt" 2>/dev/null || true
fi
[[ "$DECISION" == "CONTINUE" ]] || break
done
echo "LOOP_RESULT run_id=$RUN_ID final=$FINAL last_step=$step tokens=$CUM_TOKENS cost=$COST_ACC"
exit "$RC"
AINative_OKR_CASAN5/.specify/scripts/bash/loop-trace.py
+263
View File
@@ -0,0 +1,263 @@
#!/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())
AINative_OKR_CASAN5/.specify/scripts/bash/loop_common.py
+417
View File
@@ -0,0 +1,417 @@
"""CASAN loop-engineering shared helpers (Plan-17, harness-owned).
Deny-by-default, fail-closed primitives shared by the loop primitives
(`loop-governor.py`, `loop-convergence.py`, `loop-gate.py`). No third-party
dependency is required to *function safely*: if PyYAML is unavailable the loop
still runs under the strictest built-in ceiling (fail-closed) — never an
"unlimited" fallback.
Distinction (Plan-17 T1 17.1/17.2):
* policy file ABSENT -> use STRICT_CEILING (evaluate normally).
* policy file PRESENT but unreadable/corrupt -> raise PolicyError (caller HALTs).
State is written under a redirectable, tenant-aware root so it never pollutes the
repo (tests set CASAN_LOOP_STATE_ROOT / CASAN_TENANT_STATE_ROOT to a tmp dir).
"""
import hashlib
import json
import os
import re
import subprocess
from datetime import datetime, timezone
# Strictest possible ceiling. Used when no policy file exists or no rule matches
# a run (deny-by-default: absence of an explicit grant means the tightest budget,
# not "infinite").
STRICT_CEILING = {
"max_steps": 3,
"max_tokens": 8000,
"max_wall_clock_sec": 120,
"max_cost_usd": 0.10,
"max_corrections_per_step": 1,
"on_exceed": "halt", # halt | escalate
}
_BUDGET_KEYS = (
"max_steps",
"max_tokens",
"max_wall_clock_sec",
"max_cost_usd",
"max_corrections_per_step",
)
# Strictest convergence thresholds (Plan-17 T2). Small windows => detect a stuck /
# oscillating loop *sooner* when no policy grants a looser window (deny-by-default).
STRICT_CONVERGENCE = {
"oscillation_repeat": 3, # N identical consecutive actions => OSCILLATING
"thrash_window": 4, # A,B,A,B... over this many steps => OSCILLATING
"no_progress_window": 3, # W steps with zero forward progress => STALLED
"on_stall": "escalate", # escalate | halt
}
_CONVERGENCE_INT_KEYS = ("oscillation_repeat", "thrash_window", "no_progress_window")
class PolicyError(Exception):
"""A policy file exists but cannot be trusted (unreadable / malformed).
Callers must treat this as fail-closed (HALT), never fall back to open."""
def project_root() -> str:
return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
def now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def profile() -> str:
return os.environ.get("CASAN_PROFILE", "dev").strip() or "dev"
def git_commit() -> str:
try:
r = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=project_root(),
capture_output=True,
text=True,
timeout=5,
)
if r.returncode == 0:
return r.stdout.strip()
except Exception:
pass
return "unknown"
def _tenant_id():
t = os.environ.get("CASAN_TENANT_ID", "").strip()
if not t:
return None
if not re.fullmatch(r"[A-Za-z0-9_-]+", t):
# Same fail-closed contract as control-plane-settings.py (SEC-23 MT-01).
raise PolicyError("tenant_id_invalid")
return t
def state_root() -> str:
explicit = os.environ.get("CASAN_LOOP_STATE_ROOT")
if explicit:
return explicit
tenant = _tenant_id()
if tenant:
base = os.environ.get(
"CASAN_TENANT_STATE_ROOT",
os.path.join(project_root(), ".specify/state/tenants"),
)
return os.path.join(base, tenant, "loops")
return os.path.join(project_root(), ".specify/state/loops")
def run_dir(run_id: str) -> str:
if not run_id or not re.fullmatch(r"[A-Za-z0-9._-]+", run_id):
raise PolicyError("run_id_invalid")
return os.path.join(state_root(), "runs", run_id)
def provenance(source: str, artifact_path=None, verified: bool = False) -> dict:
"""Every primitive output carries this envelope (Plan-13 §8.6 data-contract)."""
return {
"source": source,
"artifact_path": artifact_path,
"commit": git_commit(),
"run_at": now_iso(),
"verified": bool(verified),
}
def policy_path() -> str:
explicit = os.environ.get("CASAN_LOOP_POLICY_FILE")
if explicit:
return explicit
return os.path.join(project_root(), ".specify/config/loop-policy.yaml")
def load_policy():
"""Return the parsed policy dict, or None when no policy file exists.
Fail-closed: a present-but-unreadable/malformed policy raises PolicyError so
the caller HALTs rather than silently running unbounded.
"""
path = policy_path()
if not os.path.isfile(path):
return None
try:
import yaml # optional dependency
except ImportError as exc: # cannot parse a policy we were told to honour
raise PolicyError("pyyaml_unavailable") from exc
try:
with open(path, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
except Exception as exc: # malformed YAML
raise PolicyError(f"policy_unreadable:{exc}") from exc
if data is None:
raise PolicyError("policy_empty")
if not isinstance(data, dict):
raise PolicyError("policy_not_mapping")
return data
def _coerce_budget(raw, base):
"""Overlay only the known, well-typed budget keys from `raw` onto `base`.
Unknown keys are ignored; a wrong-typed value fails closed."""
out = dict(base)
if not isinstance(raw, dict):
return out
for k in _BUDGET_KEYS:
if k in raw and raw[k] is not None:
v = raw[k]
if not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0:
raise PolicyError(f"bad_budget_value:{k}")
out[k] = v
if "on_exceed" in raw and raw["on_exceed"] is not None:
oe = raw["on_exceed"]
if oe not in ("halt", "escalate"):
raise PolicyError(f"bad_on_exceed:{oe}")
out["on_exceed"] = oe
return out
# --- governed override layer (Plan-17 T5 meta-loop) -------------------------
# A versioned, approved change recorded in the control-plane settings store can
# tighten/loosen the effective ceiling. This is how a governed meta-loop decision
# actually changes the governor's behaviour (not just a proposal on paper).
_GOV_BUDGET_MAP = {
"loop.max_steps": "max_steps",
"loop.max_tokens": "max_tokens",
"loop.max_wall_clock_sec": "max_wall_clock_sec",
"loop.max_cost_usd": "max_cost_usd",
"loop.max_corrections_per_step": "max_corrections_per_step",
}
_GOV_CONV_MAP = {
"loop.oscillation_repeat": "oscillation_repeat",
"loop.no_progress_window": "no_progress_window",
}
def cp_store_path() -> str:
"""Resolve the control-plane settings store the same way control-plane-settings.py
does, so governed loop overrides are read from exactly where they were written."""
explicit = os.environ.get("CASAN_CP_STORE_FILE")
if explicit:
return explicit
tenant = _tenant_id()
if tenant:
base = os.environ.get(
"CASAN_TENANT_STATE_ROOT",
os.path.join(project_root(), ".specify/state/tenants"),
)
return os.path.join(base, tenant, "control-plane", "settings.json")
return os.path.join(project_root(), ".specify/level5/control-plane-settings.json")
def load_governed_overrides() -> dict:
"""Best-effort read of `loop.*` governed settings. Absent/unreadable store =>
{} (no override => the YAML/strict ceiling stands, which is already safe, so a
read failure never *loosens* anything)."""
path = cp_store_path()
if not os.path.isfile(path):
return {}
try:
data = json.load(open(path, encoding="utf-8"))
settings = data.get("settings", {})
out = {}
for k, v in settings.items():
if k.startswith("loop.") and isinstance(v, dict) and "value" in v:
out[k] = v["value"]
return out
except (OSError, ValueError, KeyError, TypeError):
return {}
def org_ceiling(policy) -> dict:
"""Organization hard cap (17.19): a governed loosen can never exceed these,
even with approval. Malformed => fail-closed (PolicyError via _coerce_budget)."""
if not policy:
return {}
oc = policy.get("org_ceiling")
if not isinstance(oc, dict):
return {}
return _coerce_budget(oc, {})
def _apply_governed_budget(ceiling, policy):
overrides = load_governed_overrides()
if not overrides:
return ceiling, ""
oc = org_ceiling(policy)
applied = []
for gk, bk in _GOV_BUDGET_MAP.items():
v = overrides.get(gk)
if isinstance(v, (int, float)) and not isinstance(v, bool) and v >= 0:
if bk in oc:
v = min(v, oc[bk]) # clamp to org hard cap (defense-in-depth)
ceiling[bk] = v
applied.append(bk)
return ceiling, ("+governed(" + ",".join(applied) + ")" if applied else "")
def _apply_governed_convergence(cfg):
overrides = load_governed_overrides()
if not overrides:
return cfg, ""
applied = []
for gk, ck in _GOV_CONV_MAP.items():
v = overrides.get(gk)
if isinstance(v, int) and not isinstance(v, bool) and v >= 1:
cfg[ck] = v
applied.append(ck)
return cfg, ("+governed(" + ",".join(applied) + ")" if applied else "")
def resolve_budget(policy, prof: str, delegation_level=None, project=None):
"""Compute the effective ceiling for a run.
Deny-by-default: start from STRICT_CEILING and overlay, in order,
profile.defaults -> delegation_levels[L] -> projects[name] -> governed
overrides (clamped to org_ceiling). Any field not granted stays strictest.
Returns (ceiling_dict, source_tag).
"""
ceiling = dict(STRICT_CEILING)
sources = []
if policy:
profiles = policy.get("profiles")
if not isinstance(profiles, dict):
raise PolicyError("policy_missing_profiles")
prof_block = profiles.get(prof)
if isinstance(prof_block, dict):
defaults = prof_block.get("defaults")
if isinstance(defaults, dict):
ceiling = _coerce_budget(defaults, ceiling)
sources.append(f"{prof}.defaults")
if delegation_level:
levels = prof_block.get("delegation_levels")
if isinstance(levels, dict) and delegation_level in levels:
ceiling = _coerce_budget(levels[delegation_level], ceiling)
sources.append(f"delegation:{delegation_level}")
if project:
projects = prof_block.get("projects")
if isinstance(projects, dict) and project in projects:
ceiling = _coerce_budget(projects[project], ceiling)
sources.append(f"project:{project}")
else:
# Unknown profile => no matching rule => strictest (deny-by-default).
sources.append(f"strict-default(no-profile:{prof})")
else:
sources.append("strict-default(no-policy)")
ceiling, gov = _apply_governed_budget(ceiling, policy)
tag = "+".join(sources) if sources else f"strict-default(empty:{prof})"
return ceiling, tag + gov
def _coerce_convergence(raw, base):
out = dict(base)
if not isinstance(raw, dict):
return out
for k in _CONVERGENCE_INT_KEYS:
if k in raw and raw[k] is not None:
v = raw[k]
if not isinstance(v, int) or isinstance(v, bool) or v < 1:
raise PolicyError(f"bad_convergence_value:{k}")
out[k] = v
if "on_stall" in raw and raw["on_stall"] is not None:
os_ = raw["on_stall"]
if os_ not in ("halt", "escalate"):
raise PolicyError(f"bad_on_stall:{os_}")
out["on_stall"] = os_
return out
def resolve_convergence(policy, prof: str):
"""Effective convergence thresholds for a profile. Deny-by-default: absent
policy / profile => strictest (detect stalls soonest). A governed override
layer (meta-loop) can adjust the windows. Returns (dict, tag)."""
cfg = dict(STRICT_CONVERGENCE)
if policy:
profiles = policy.get("profiles")
if not isinstance(profiles, dict):
raise PolicyError("policy_missing_profiles")
prof_block = profiles.get(prof)
if isinstance(prof_block, dict):
raw = prof_block.get("convergence")
if isinstance(raw, dict):
cfg = _coerce_convergence(raw, cfg)
src = f"{prof}.convergence"
else:
src = f"strict-default(no-convergence:{prof})"
else:
src = f"strict-default(no-profile:{prof})"
else:
src = "strict-default(no-policy)"
cfg, gov = _apply_governed_convergence(cfg)
return cfg, src + gov
# ---------------------------------------------------------------------------
# Hash-linked loop audit log (self-contained tamper-evidence). Shares the same
# canonical-JSON + SHA-256 scheme as control-plane-settings.py so a future
# unified verifier (Plan-17 T4 / sync-point S2) can adopt it unchanged.
# ---------------------------------------------------------------------------
GENESIS_HASH = "0" * 64
def _canon(entry) -> str:
return json.dumps(entry, sort_keys=True, ensure_ascii=False)
def _hash_entry(entry) -> str:
return hashlib.sha256(_canon(entry).encode("utf-8")).hexdigest()
# Public serializer/verifier (sync-point S2: one serializer, one verifier). The
# per-run loop trace (Track 4) reuses these so its hash-chain is byte-compatible
# with the audit log and any future unified verifier.
def canonical(entry) -> str:
return _canon(entry)
def chain_hash(entry) -> str:
return _hash_entry(entry)
def audit_log_path() -> str:
return os.path.join(state_root(), "audit", "loop-audit.jsonl")
def append_audit(event: dict) -> dict:
"""Append a hash-linked audit record. Append-only; each record chains to the
previous via prev_hash so any later edit breaks the chain."""
path = audit_log_path()
os.makedirs(os.path.dirname(path), exist_ok=True)
prev = GENESIS_HASH
seq = 0
if os.path.isfile(path):
with open(path, encoding="utf-8") as fh:
last = None
for line in fh:
line = line.strip()
if line:
last = line
seq += 1
if last:
try:
prev = json.loads(last).get("hash", GENESIS_HASH)
except ValueError:
prev = GENESIS_HASH
base = {
"seq": seq,
"ts": now_iso(),
"prev_hash": prev,
"event": event,
}
base["hash"] = _hash_entry(base)
with open(path, "a", encoding="utf-8") as fh:
fh.write(_canon(base) + "\n")
return base