erp/fleet/harness/promote/pipeline.py

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