#!/usr/bin/env python3 """Gated promote pipeline: rehearse → judge → human gate → apply → judge. Stages are chained by artefacts on disk, not by discipline: each one refuses to run until the previous produced its file and that file says what this stage needs. See README.md for why. Stdlib only. """ from __future__ import annotations import argparse import getpass import hashlib import json import os import subprocess import sys import urllib.error import urllib.request from datetime import datetime, timezone REPO = os.path.realpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..")) SKILL = os.path.join(REPO, ".claude", "skills") SANDBOX_HOST = "erp-sandbox.arcodange.lab" STAGES = {"rehearsal": "01-rehearsal.json", "pre": "02-pre-verdict.json", "gate": "03-gate.json", "applied": "04-applied.json", "post": "05-post-verdict.json"} def die(msg: str) -> None: sys.exit(f"pipeline: REFUSED — {msg}") def now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def digest(obj) -> str: return hashlib.sha256(json.dumps(obj, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:16] def art(run_dir: str, stage: str) -> str: return os.path.join(run_dir, STAGES[stage]) def read_stage(run_dir: str, stage: str, why: str) -> dict: p = art(run_dir, stage) if not os.path.exists(p): die(f"{why}\n missing: {p}\n run the earlier stage first.") return json.load(open(p)) def write_stage(run_dir: str, stage: str, payload: dict) -> str: p = art(run_dir, stage) json.dump(payload, open(p, "w"), indent=2, ensure_ascii=False) with open(os.path.join(run_dir, "journal.jsonl"), "a") as j: j.write(json.dumps({"at": now(), "stage": stage, "file": os.path.basename(p)}, ensure_ascii=False) + "\n") return p def env_from(path: str) -> dict: cfg = {} for line in open(path): if "=" in line and not line.strip().startswith("#"): k, v = line.strip().split("=", 1) cfg[k] = v.strip().strip('"').strip("'") return cfg def api(base: str, key: str, method: str, path: str, body=None): url = f"{base.rstrip('/')}/api/index.php{path}" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={"DOLAPIKEY": key, "Content-Type": "application/json", "Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=120) as r: raw = r.read().decode() except urllib.error.HTTPError as e: return {"_error": e.code, "_body": e.read().decode()[:300]} try: return json.loads(raw) except json.JSONDecodeError: return raw.strip().strip('"') # --- stage 1: rehearse ------------------------------------------------------- def stage_rehearse(args) -> None: manifest = json.load(open(args.manifest)) os.makedirs(args.run_dir, exist_ok=True) cfg = env_from(os.path.join(SKILL, "dolibarr-sandbox-write", ".env")) # the write skill namespaces its keys (DOLIBARR_SANDBOX_*), the read skill does not base = cfg.get("DOLIBARR_SANDBOX_URL") or cfg.get("DOLIBARR_URL", "") skey = cfg.get("DOLIBARR_SANDBOX_API_KEY") or cfg.get("DOLIBARR_API_KEY", "") if SANDBOX_HOST not in base: die(f"the write .env points at {base!r}, not the sandbox — rehearsal must target {SANDBOX_HOST}") ops, before = manifest.get("ops", []), {} for probe in manifest.get("observe", []): before[probe] = api(base, skey, "GET", probe) results = [] for op in ops: call = op.get("api") if not call: die(f"op {op.get('label')!r} has no 'api' block. An op is defined ONCE and replayed on both\n" " targets — two descriptions of the same write are two things that can disagree.") r = api(base, skey, call["method"], call["path"], call.get("body")) failed = isinstance(r, dict) and "_error" in r results.append({"label": op.get("label", ""), "method": call["method"], "path": call["path"], "rc": 1 if failed else 0, "result": r}) print(f" [{op.get('label','op')}] {call['method']} {call['path']} -> {'ok' if not failed else r}") for follow in op.get("then", []): fr = api(base, skey, follow["method"], follow["path"].replace("{id}", str(r)), follow.get("body")) ffailed = isinstance(fr, dict) and "_error" in fr results.append({"label": f"{op.get('label','')} :: {follow.get('label','follow-up')}", "method": follow["method"], "path": follow["path"], "rc": 1 if ffailed else 0, "result": fr}) print(f" └ {follow.get('label','follow-up')} -> {'ok' if not ffailed else fr}") after = {p: api(base, skey, "GET", p) for p in manifest.get("observe", [])} ok = all(r["rc"] == 0 for r in results) payload = {"at": now(), "target": base, "manifest_file": os.path.abspath(args.manifest), "manifest_digest": digest(manifest), "manifest": manifest, "results": results, "observed_before": before, "observed_after": after, "all_writes_succeeded": ok} print(f"→ {write_stage(args.run_dir, 'rehearsal', payload)} (writes ok: {ok})") if not ok: print(" some ops failed — the pre-gate judge will see that, and apply stays blocked.") # --- stage 2 & 5: judges ----------------------------------------------------- def call_runtime(runtime: str, prompt: str, model: str | None) -> tuple[str, str]: if runtime == "mistral": r = subprocess.run(["vibe", "-p", prompt, "--max-turns", "1", "--output", "text"], capture_output=True, text=True, timeout=600) if r.returncode != 0: die(f"vibe failed: {r.stderr[-300:]}") return r.stdout.strip(), "vibe -p (mistral)" if runtime in ("ornith", "mlx"): model = model or "leonsarmiento/Ornith-1.0-35B-5bit-mlx" body = json.dumps({"model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0, "max_tokens": 3000}).encode() req = urllib.request.Request("http://127.0.0.1:18080/v1/chat/completions", data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=900) as r: m = json.load(r)["choices"][0]["message"] return (m.get("content") or m.get("reasoning") or ""), model die(f"unknown runtime {runtime!r} (admitted: mistral, ornith, mlx)") raise AssertionError def stage_judge(args) -> None: here = os.path.dirname(os.path.abspath(__file__)) if args.stage == "pre": reh = read_stage(args.run_dir, "rehearsal", "a pre-gate judge needs a rehearsal to judge") tmpl = open(os.path.join(here, "judges", "pre-gate.md")).read() evidence = {k: reh[k] for k in ("manifest", "results", "observed_before", "observed_after", "all_writes_succeeded")} else: applied = read_stage(args.run_dir, "applied", "a post-gate judge needs a production apply to verify") reh = read_stage(args.run_dir, "rehearsal", "the post-gate judge compares production to the rehearsal") tmpl = open(os.path.join(here, "judges", "post-gate.md")).read() evidence = {"manifest": reh["manifest"], "sandbox_after": reh["observed_after"], "production_after": applied.get("observed_after"), "apply_results": applied.get("results")} prompt = f"{tmpl}\n\n--- EVIDENCE (JSON) ---\n{json.dumps(evidence, indent=2, ensure_ascii=False)}\n--- END ---" text, model = call_runtime(args.runtime, prompt, args.model) verdict = "BLOCK" if "BLOCK" in text.upper()[:400] else ("PASS" if "PASS" in text.upper()[:400] else "UNCLEAR") payload = {"at": now(), "stage": args.stage, "runtime": args.runtime, "model": model, "prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest(), "verdict": verdict, "response": text} print(f"→ {write_stage(args.run_dir, args.stage, payload)} verdict={verdict} ({model})") print(text[:1200]) # --- stage 3: the human gate ------------------------------------------------- def stage_gate(args) -> None: reh = read_stage(args.run_dir, "rehearsal", "nothing to approve — rehearse first") pre = read_stage(args.run_dir, "pre", "the human decides WITH a judge's opinion, not without one") m = reh["manifest"] print("=" * 72) print("HUMAN GATE — this approves writing to the PRODUCTION ledger.") print("=" * 72) print(f"change-set : {m.get('title', '(untitled)')}") print(f"digest : {reh['manifest_digest']} (approval binds to this exact change-set)") print(f"ops : {len(m.get('ops', []))}") for op in m.get("ops", []): print(f" - {op.get('label') or op['script']}") print(f"rehearsal : writes ok = {reh['all_writes_succeeded']} on {reh['target']}") print(f"judge ({pre['runtime']}) : {pre['verdict']}") for line in pre["response"].strip().splitlines()[:8]: print(f" | {line[:100]}") print("=" * 72) if not reh["all_writes_succeeded"]: print("NOTE: the rehearsal had failing ops. Approving anyway is your call, and is recorded.") if pre["verdict"] == "BLOCK": print("NOTE: the judge says BLOCK. You may still approve — the override is recorded.") if args.decision: decision, who = args.decision, args.by or getpass.getuser() else: decision = input("\ntype 'approve' to promote, anything else to abort: ").strip().lower() who = input("your name (recorded in the evidence pack): ").strip() or getpass.getuser() payload = {"at": now(), "decision": "approved" if decision == "approve" else "rejected", "by": who, "manifest_digest": reh["manifest_digest"], "judge_verdict": pre["verdict"], "override_of_judge": pre["verdict"] == "BLOCK" and decision == "approve", "override_of_failed_rehearsal": (not reh["all_writes_succeeded"]) and decision == "approve"} print(f"→ {write_stage(args.run_dir, 'gate', payload)} decision={payload['decision']} by {who}") # --- stage 4: apply to production ------------------------------------------- def stage_apply(args) -> None: reh = read_stage(args.run_dir, "rehearsal", "nothing rehearsed") gate = read_stage(args.run_dir, "gate", "production requires a human decision on record") if gate["decision"] != "approved": die(f"the gate recorded '{gate['decision']}' by {gate['by']} — not approved.") if gate["manifest_digest"] != reh["manifest_digest"]: die("the approved change-set is not the one about to be applied " f"(approved {gate['manifest_digest']}, current {reh['manifest_digest']}).\n" " Re-run the gate on the current change-set.") if os.environ.get("ARCO_PROD_CONFIRM") != "I-UNDERSTAND-THIS-WRITES-PROD": die("set ARCO_PROD_CONFIRM=I-UNDERSTAND-THIS-WRITES-PROD to apply to production") # The production writer is its OWN user with the narrow `prod-write` scope # (test/scopes.ts) — not the read-only `ai_agent` whose key every read skill # holds. A read credential must never be able to write the ledger, and the # writer must not be lying around in a file a dozen skills load. cfg = env_from(os.path.join(SKILL, "dolibarr", ".env")) base = cfg["DOLIBARR_URL"] key_file = os.path.join(REPO, "test", ".ai_agent_prod_prod_write.key") if not os.path.exists(key_file): die("no production writer key.\n" f" expected: {key_file}\n" " provision it: DOLIBARR_ADDRESS=https://erp.arcodange.lab " "ARCO_ALLOW_PRODUCTION=erp.arcodange.lab " "ARCO_PROD_CONFIRM=I-UNDERSTAND-THIS-WRITES-PROD \\\n" " deno run -A test/provisionAiUser.ts --scope prod-write --env production") key = open(key_file).read().strip() if SANDBOX_HOST in base: die(f"the prod .env points at the sandbox ({base}) — nothing to promote to") print(f"*** PRODUCTION: {base} — approved by {gate['by']} at {gate['at']} ***") m = reh["manifest"] results = [] for op in m.get("ops", []): call = op["api"] r = api(base, key, call["method"], call["path"], call.get("body")) failed = isinstance(r, dict) and "_error" in r results.append({"label": op.get("label"), "method": call["method"], "path": call["path"], "result": r, "ok": not failed}) print(f" [{op.get('label')}] {call['method']} {call['path']} -> {'ok' if not failed else r}") if failed and not args.keep_going: break for follow in op.get("then", []): fr = api(base, key, follow["method"], follow["path"].replace("{id}", str(r)), follow.get("body")) ffailed = isinstance(fr, dict) and "_error" in fr results.append({"label": f"{op.get('label')} :: {follow.get('label','follow-up')}", "method": follow["method"], "path": follow["path"], "result": fr, "ok": not ffailed}) print(f" └ {follow.get('label','follow-up')} -> {'ok' if not ffailed else fr}") if ffailed and not args.keep_going: break after = {p: api(base, key, "GET", p) for p in m.get("observe", [])} payload = {"at": now(), "target": base, "approved_by": gate["by"], "manifest_digest": reh["manifest_digest"], "results": results, "observed_after": after, "all_ok": all(r["ok"] for r in results)} print(f"→ {write_stage(args.run_dir, 'applied', payload)} (all ok: {payload['all_ok']})") def main() -> int: ap = argparse.ArgumentParser(description="gated promote pipeline") sub = ap.add_subparsers(dest="cmd", required=True) r = sub.add_parser("rehearse"); r.add_argument("--manifest", required=True); r.add_argument("--run-dir", required=True) j = sub.add_parser("judge"); j.add_argument("--run-dir", required=True) j.add_argument("--stage", choices=["pre", "post"], required=True) j.add_argument("--runtime", default="mistral"); j.add_argument("--model") g = sub.add_parser("gate"); g.add_argument("--run-dir", required=True) g.add_argument("--decision", choices=["approve", "reject"]); g.add_argument("--by") a = sub.add_parser("apply"); a.add_argument("--run-dir", required=True); a.add_argument("--keep-going", action="store_true") args = ap.parse_args() {"rehearse": stage_rehearse, "judge": stage_judge, "gate": stage_gate, "apply": stage_apply}[args.cmd](args) return 0 if __name__ == "__main__": sys.exit(main())