"""CAIN-A75-038 offline verifier (rules embedded verbatim). CAIN-A75-038 parallel-run rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier.

A supervisor runs several coding agents on desktops at once. The review recomputes whether parallel work was isolated and
merged honestly:

  concurrency  the number of simultaneous runs is within the supervisor's policy
  workspaces   runs have distinct workspaces when the policy requires isolation
  writes       two simultaneous runs never write the same path unless a live lock covers it
  resources    each run stays inside its CPU/memory envelope
  merge        every merge conflict is resolved and the merged artifact carries a hash
  outcomes     every run has a status; a failed run wrote nothing that reached the merge
  claims       every boolean the record asserts matches what this review computed

Verdict: SOUND, else UNSOUND with every failing check named. Malformed input is UNSOUND.
It reviews the run record; it does not supervise the desktops.
"""
from __future__ import annotations

import math
import re

_ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$")
_HEX = re.compile(r"^[0-9a-f]{64}$")


class Bad(ValueError):
    pass


def _num(v, what, lo=0.0, hi=1e12):
    if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or not (lo <= v <= hi):
        raise Bad(f"{what} must be a finite number in [{lo}, {hi}]")
    return float(v)


def evaluate(record):
    checks, summary = [], {"runs": {}, "conflicts": []}

    def c(cid, ok, detail=""):
        checks.append({"id": cid, "ok": bool(ok), "detail": detail})

    try:
        if not isinstance(record, dict) or record.get("schema") != "cain42.a75.parallel.runs.v1":
            raise Bad("not a cain42.a75.parallel.runs.v1 record")
        sup, pol = record.get("supervisor"), record.get("policy")
        if not all(isinstance(x, dict) for x in (sup, pol)):
            raise Bad("supervisor and policy must be objects")
        max_conc = pol.get("max_concurrency")
        if isinstance(max_conc, bool) or not isinstance(max_conc, int) or max_conc < 1:
            raise Bad("policy.max_concurrency must be a positive integer")
        isolate = pol.get("require_isolated_workspaces") is True
        forbid_same = pol.get("forbid_same_file_writes") is True

        runs = record.get("runs")
        if not isinstance(runs, list) or not runs:
            raise Bad("runs must be a non-empty list")
        R = {}
        for r in runs:
            if not isinstance(r, dict) or not isinstance(r.get("id"), str) or not _ID.match(r["id"]):
                raise Bad("a run needs an id")
            if r["id"] in R:
                raise Bad(f"duplicate run {r['id']}")
            R[r["id"]] = r
            c(f"{r['id']}.status", r.get("status") in ("completed", "failed", "cancelled"), f"status {r.get('status')!r}")
            c(f"{r['id']}.window", _num(r.get("finished_at"), f"{r['id']}.finished_at") >= _num(r.get("started_at"), f"{r['id']}.started_at"),
              "finish is after start")
            c(f"{r['id']}.workspace", isinstance(r.get("workspace"), str) and bool(r["workspace"]), f"workspace {r.get('workspace')!r}")
            c(f"{r['id']}.cpu", _num(r.get("cpu_pct"), f"{r['id']}.cpu_pct", 0.0, 100.0) <= 100.0, "CPU within 100%")
            c(f"{r['id']}.memory", _num(r.get("memory_mb"), f"{r['id']}.memory_mb", 0.0) <= _num(pol.get("max_memory_mb"), "policy.max_memory_mb", 1.0),
              "memory within the policy")
            writes = r.get("writes") if isinstance(r.get("writes"), list) else []
            for w in writes:
                if not isinstance(w, dict) or not isinstance(w.get("path"), str):
                    raise Bad(f"{r['id']}: a write needs a path")
                c(f"{r['id']}.write.{w['path']}", isinstance(w.get("sha256"), str) and bool(_HEX.match(w["sha256"])),
                  "a write carries its content hash")
            summary["runs"][r["id"]] = {"workspace": r.get("workspace"), "status": r.get("status"),
                                        "started_at": r.get("started_at"), "finished_at": r.get("finished_at"),
                                        "writes": [w.get("path") for w in writes]}

        peak = max(sum(1 for rr in runs if rr["started_at"] <= r["started_at"] < rr["finished_at"]) or 1 for r in runs)
        c("concurrency.limit", peak <= max_conc, f"peak {peak} <= {max_conc}")

        if isolate:
            workspaces = [r.get("workspace") for r in runs]
            c("workspaces.isolated", len(set(workspaces)) == len(workspaces), f"workspaces {workspaces}")

        locks = record.get("locks") if isinstance(record.get("locks"), list) else []
        locked = {}
        for lk in locks:
            if not isinstance(lk, dict) or not isinstance(lk.get("resource"), str):
                raise Bad("a lock needs a resource")
            live = _num(lk.get("expires_at"), f"lock.{lk['resource']}.expires_at") > 0
            if live:
                locked.setdefault(lk["resource"], []).append(lk.get("holder"))
        if forbid_same:
            path_writers = {}
            for r in runs:
                for w in r.get("writes") or []:
                    path_writers.setdefault(w["path"], set()).add(r["id"])
            overlap = {p: sorted(v) for p, v in path_writers.items() if len(v) > 1 and p not in locked}
            c("writes.no_concurrent_same_path", not overlap, f"overlapping writers: {overlap}" if overlap else "no overlapping writes")

        for lk in locks:
            if isinstance(lk, dict):
                c(f"lock.{lk.get('resource')}.holder_exists", lk.get("holder") in R, f"holder {lk.get('holder')!r}")

        merge = record.get("merge") if isinstance(record.get("merge"), dict) else {}
        conflicts = merge.get("conflicts") if isinstance(merge.get("conflicts"), list) else []
        for k, mc in enumerate(conflicts):
            if not isinstance(mc, dict):
                raise Bad("a merge conflict must be an object")
            c(f"merge.{k}.resolved", isinstance(mc.get("resolution"), str) and bool(mc["resolution"].strip()), "conflict resolved")
            c(f"merge.{k}.runs", isinstance(mc.get("runs"), list) and all(x in R for x in mc["runs"]), f"runs {mc.get('runs')!r}")
        c("merge.artifact_hash", isinstance(merge.get("merged_sha256"), str) and bool(_HEX.match(merge["merged_sha256"])),
          "the merged artifact is hashed")
        failed = [r["id"] for r in runs if r.get("status") == "failed"]
        merged_paths = set(merge.get("paths") or [])
        leaked = sorted(p for rid in failed for p in (summary["runs"][rid]["writes"] or []) if p in merged_paths)
        c("outcomes.failed_not_merged", not leaked, f"failed-run writes reached the merge: {leaked}" if leaked else "no failed-run write merged")

        claims = record.get("claims") if isinstance(record.get("claims"), dict) else {}
        computed = {
            "within_concurrency": peak <= max_conc,
            "isolated_workspaces": (not isolate) or len({r.get("workspace") for r in runs}) == len(runs),
            "merge_resolved": all(isinstance(mc.get("resolution"), str) and bool(mc["resolution"].strip()) for mc in conflicts),
        }
        for key in sorted(claims):
            c(f"claims.{key}", key in computed and claims[key] == computed[key], f"claimed {claims[key]!r}, computed {computed.get(key)!r}")
        summary["computed"] = computed
        summary["peak_concurrency"] = peak
    except Bad as e:
        checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)})
        return "UNSOUND", checks, summary
    except Exception as e:  # noqa: BLE001
        checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]})
        return "UNSOUND", checks, summary
    return ("SOUND" if all(x["ok"] for x in checks) else "UNSOUND"), checks, summary


# ---------------------------------------------------------------------------- offline verifier (no CAIN imports)
# Every name below is prefixed _kit_verifier_ / _b64 / _hl / _json / _sys so it cannot shadow a name in the rules above.
import base64 as _b64, hashlib as _hl, json as _json, sys as _sys
_KIT_VERIFIER_GENESIS = _hl.sha256(b"cain42.a75.parallel.genesis").hexdigest()


def _kit_verifier_canon(o):
    return _json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()


def _kit_verifier_sha(o):
    return _hl.sha256(_kit_verifier_canon(o)).hexdigest()


def _kit_verifier_main(argv):
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
    if len(argv) < 2:
        print("usage: verify_offline.py run.json --pinned-key BASE64"); return 2
    run = _json.load(open(argv[1]))
    pin = argv[argv.index("--pinned-key") + 1] if "--pinned-key" in argv else None
    problems, prev, recs = [], _KIT_VERIFIER_GENESIS, run.get("records") or []
    for i, r in enumerate(recs):
        body = {k: v for k, v in r.items() if k not in ("record_hash", "signature_b64")}
        if _kit_verifier_sha(body) != r.get("record_hash"): problems.append(f"record {i}: record_hash does not match")
        if body.get("subject_sha256") != _kit_verifier_sha(body.get("subject")): problems.append(f"record {i}: subject_sha256 does not match")
        if pin and r.get("public_key_b64") != pin: problems.append(f"record {i}: different key than the pinned one")
        try:
            Ed25519PublicKey.from_public_bytes(_b64.b64decode(pin or r.get("public_key_b64"))).verify(
                _b64.b64decode(r.get("signature_b64", "")), str(r.get("record_hash")).encode())
        except Exception:
            problems.append(f"record {i}: signature does not verify")
        if r.get("prev_record_hash") != prev: problems.append(f"record {i}: chain link broken")
        v, checks, _ = evaluate(r.get("subject"))
        if v != r.get("verdict") or checks != r.get("checks"):
            problems.append(f"record {i}: re-running the rules gives {v}, the record says {r.get('verdict')}")
        prev = r.get("record_hash")
    if run.get("run_hash") != _kit_verifier_sha([r.get("record_hash") for r in recs]): problems.append("run_hash does not match")
    for p in problems: print("  -", p)
    if not pin: print("WARNING: no --pinned-key given; the signature was checked against the key inside the file")
    print("VERDICT:", "VERIFIED" if not problems else "TAMPERED OR INVALID", f"(CAIN-A75-038, {len(recs)} record(s), verdicts {run.get('verdicts')})")
    return 0 if not problems else 1


if __name__ == "__main__":
    _sys.exit(_kit_verifier_main(_sys.argv))