scripts/forge.py

#!/usr/bin/env python3
"""Forge a REAL trained surrogate for szl-provctl.
Kernel = ground truth. Surrogate = fast provenance-DAG anomaly classifier: given
graph-structural observables of a provenance DAG (node/edge counts, indegrees,
per-node chain-integrity recompute, edge-head-match recompute, orphan/cycle
signals), predict which anomaly class the kernel's `verify_dag(...)` would report.

The kernel itself is the labeler: `ProvenanceDAG.verify_dag()['first_break']['kind']`
maps to one of {clean, cycle, edge-mismatch, missing-dependency, chain-integrity}.
Each corrupted DAG injects EXACTLY one anomaly (single-aspect rule). A sample of
labels is re-audited by full kernel replay and MUST agree.
Seeded, receipted, reproducible."""
import json, os, random, sys, time, hashlib, platform
_here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if os.path.isdir(os.path.join(_here, "build", "torch-universal")):
    sys.path.insert(0, os.path.join(_here, "build", "torch-universal"))  # in-repo run
else:
    sys.path.insert(0, "/tmp/kernel-probe/szl-provctl/build/torch-universal")  # forge-dev run
import szl_provctl as pc
from szl_provctl._chain import UnifiedReceiptChain, GENESIS
from szl_provctl._dag import ProvenanceDAG, ProvenanceNode
import numpy as np
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, recall_score
import joblib

SEED = 20260721
random.seed(SEED); np.random.seed(SEED)
T0 = time.time()

# kernel first_break kinds -> our class names.
# NOTE (measured kernel limitation, honest): the kernel's `missing_dependency`
# verdict is UNREACHABLE — `_topo_order()` raises KeyError on any ghost edge
# BEFORE verify_dag's per-edge check runs, so no DAG can be labeled
# `missing_dependency` through the public API. We therefore do NOT include that
# class (we never fabricate a code path the kernel cannot actually produce).
KIND_TO_CLASS = {
    None: "clean",
    "cycle": "cycle",
    "edge_mismatch": "edge-mismatch",
    "chain_integrity": "chain-integrity",
}
CLASSES = ["clean", "cycle", "edge-mismatch", "chain-integrity"]


def make_chain(n_receipts):
    ch = UnifiedReceiptChain()
    for i in range(n_receipts):
        ch.emit(random.choice(["governed_norm", "lambda_gate", "govsign"]),
                random.choice(["rms_norm", "layer_norm", "gate_decision"]),
                {"seq_local": i, "eps": random.choice([1e-6, 1e-5]), "val": random.random()})
    return ch


def build_clean_dag(n_nodes):
    """A valid multi-run DAG: each node may depend on earlier nodes; heads asserted correctly."""
    dag = ProvenanceDAG()
    ids = [f"run{i}" for i in range(n_nodes)]
    for i, nid in enumerate(ids):
        ch = make_chain(random.randint(1, 4))
        deps = []
        if i > 0 and random.random() < 0.7:
            deps = random.sample(ids[:i], k=random.randint(1, min(2, i)))
        verdict = "BLOCKED" if random.random() < 0.12 else "ALLOWED"
        dag.add_run(nid, ch, deps=deps, verdict=verdict)
    return dag, ids


def corrupt(dag, ids, cls):
    """Inject EXACTLY one anomaly of type cls into a clean DAG (mutating node internals)."""
    nodes = dag._nodes
    if cls == "cycle":
        # create a 2-cycle: pick a node with a dep, add a reverse edge dep->that
        cands = [nid for nid in ids if nodes[nid].deps]
        if not cands:
            # force an edge then reverse it
            a, b = ids[0], ids[1]
            nodes[b].deps.append((a, nodes[a].head()))
            cands = [b]
        target = random.choice(cands)
        dep_id = nodes[target].deps[0][0]
        # add target as a dependency of dep_id -> cycle target<->dep_id
        nodes[dep_id].deps.append((target, nodes[target].head()))
    elif cls == "edge-mismatch":
        cands = [nid for nid in ids if nodes[nid].deps]
        if not cands:
            a, b = ids[0], ids[1]
            nodes[b].deps.append((a, nodes[a].head()))
            cands = [b]
        target = random.choice(cands)
        dep_id, _asserted = nodes[target].deps[0]
        # assert a WRONG head for the edge (single-aspect: only the asserted head changes)
        bad = hashlib.sha3_256(f"tamper{target}".encode()).hexdigest()
        nodes[target].deps[0] = (dep_id, bad)
    elif cls == "chain-integrity":
        # flip a byte inside one node's chain blob body digest (break internal chain)
        target = random.choice(ids)
        records = json.loads(nodes[target].chain_blob)
        if not records:
            ch = make_chain(2); nodes[target].chain_blob = ch.to_json()
            records = json.loads(nodes[target].chain_blob)
        j = random.randrange(len(records))
        d = bytearray(bytes.fromhex(records[j]["digest"])); d[0] ^= 0xFF
        records[j]["digest"] = bytes(d).hex()
        nodes[target].chain_blob = json.dumps(records, sort_keys=True, separators=(",", ":"))
    return dag


def features(dag):
    """Graph-structural observables + cheap per-node/edge recomputes."""
    nodes = dag._nodes
    ids = list(nodes.keys())
    n_nodes = len(ids)
    all_deps = [(nid, dep, head) for nid in ids for (dep, head) in nodes[nid].deps]
    n_edges = len(all_deps)
    indeg = {nid: len(nodes[nid].deps) for nid in ids}
    indeg_vals = list(indeg.values()) or [0]
    # missing-dependency signal: edges pointing at ids not in the DAG
    n_missing = sum(1 for (_n, dep, _h) in all_deps if dep not in nodes)
    # edge-mismatch signal: asserted head != actual head (for present deps)
    n_edge_mismatch = sum(1 for (_n, dep, h) in all_deps
                          if dep in nodes and h != nodes[dep].head())
    # chain-integrity signal: per-node internal chain verify
    n_chain_bad = 0
    for nid in ids:
        ok, _d, _b = UnifiedReceiptChain.verify_json(nodes[nid].chain_blob)
        if not ok:
            n_chain_bad += 1
    # cycle signal: detect a cycle via DFS on present-dep edges
    adj = {nid: [dep for (dep, _h) in nodes[nid].deps if dep in nodes] for nid in ids}
    has_cycle = _detect_cycle(adj)
    n_blocked = sum(1 for nid in ids if nodes[nid].verdict == "BLOCKED")
    return [
        n_nodes, n_edges,
        max(indeg_vals), float(np.mean(indeg_vals)),
        n_missing, n_edge_mismatch, n_chain_bad,
        int(has_cycle), n_blocked,
        float(n_edges) / max(n_nodes, 1),
    ]


def _detect_cycle(adj):
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {n: WHITE for n in adj}
    def dfs(u):
        color[u] = GRAY
        for v in adj.get(u, []):
            if color.get(v, WHITE) == GRAY:
                return True
            if color.get(v, WHITE) == WHITE and dfs(v):
                return True
        color[u] = BLACK
        return False
    return any(color[n] == WHITE and dfs(n) for n in adj)


FEATURE_NAMES = ["n_nodes", "n_edges", "max_indegree", "mean_indegree",
                 "n_missing_deps", "n_edge_mismatch", "n_chain_bad",
                 "has_cycle", "n_blocked_nodes", "edge_density"]


def kernel_class(dag):
    """Ground truth: run the kernel verify_dag and map first_break.kind to a class."""
    rep = pc.verify_dag(dag)
    if rep["ok"]:
        return "clean"
    kind = (rep.get("first_break") or {}).get("kind")
    return KIND_TO_CLASS.get(kind, kind if kind in CLASSES else "chain-integrity")


def make_dag(cls):
    """Build a DAG whose kernel-derived label is GUARANTEED to equal cls."""
    for _ in range(60):
        n = random.randint(3, 8)
        dag, ids = build_clean_dag(n)
        if cls != "clean":
            corrupt(dag, ids, cls)
        derived = kernel_class(dag)
        if derived == cls:
            return dag
    raise RuntimeError(f"could not synthesize a kernel-confirmed {cls} DAG")


# ---- generate ----
X, y = [], []
dags = []
PER_CLASS = {c: (2600 if c == "clean" else 1900) for c in CLASSES}
plan = []
for c, k in PER_CLASS.items():
    plan += [c] * k
random.shuffle(plan)
for cls in plan:
    dag = make_dag(cls)
    X.append(features(dag)); y.append(cls); dags.append((dag, cls))

# ground-truth audit: full kernel replay must agree
audit_checked = 0
for i in random.sample(range(len(dags)), 700):
    dag, cls = dags[i]
    assert kernel_class(dag) == cls, f"kernel disagrees DAG {i}: {kernel_class(dag)} != {cls}"
    audit_checked += 1

X = np.array(X, dtype=np.float64); y = np.array(y)
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.2, random_state=SEED, stratify=y)
clf = HistGradientBoostingClassifier(random_state=SEED, max_iter=300, early_stopping=True)
clf.fit(Xtr, ytr)
pred = clf.predict(Xte)
acc = accuracy_score(yte, pred)
per_class_recall = {c: float(recall_score(yte == c, pred == c, zero_division=0)) for c in CLASSES}

out_dir = os.path.dirname(os.path.abspath(__file__))
joblib.dump(clf, f"{out_dir}/model.joblib")
model_sha = hashlib.sha256(open(f"{out_dir}/model.joblib", "rb").read()).hexdigest()
receipt = {
  "artifact": "SZLHOLDINGS/szl-provctl surrogate v1",
  "role": "provenance-DAG anomaly classifier surrogate — kernel remains ground truth",
  "generator": {"script": "scripts/forge.py", "seed": SEED, "kernel_version": pc.__version__,
                 "kernel_labelled": True, "kernel_audited_dags": audit_checked,
                 "labeler": "ProvenanceDAG.verify_dag()['first_break']['kind']"},
  "data": {"rows": int(len(y)), "classes": CLASSES,
            "class_counts": {c: int((y == c).sum()) for c in CLASSES},
            "split": "80/20 stratified", "features": FEATURE_NAMES,
            "feature_policy": "graph-structural observables + cheap per-node chain recompute + edge-head-match recompute; the surrogate never replaces the kernel's full verify_dag traversal"},
  "model": {"type": "sklearn.HistGradientBoostingClassifier",
             "params": {"max_iter": 300, "early_stopping": True, "random_state": SEED},
             "file": "model.joblib", "sha256": model_sha},
  "metrics_MEASURED": {"fidelity_vs_kernel_heldout": round(float(acc), 4),
                        "test_accuracy": round(float(acc), 4),
                        "per_class_recall": {k: round(v, 4) for k, v in per_class_recall.items()}},
  "environment": {"python": platform.python_version(), "sklearn": __import__("sklearn").__version__,
                   "numpy": np.__version__, "host": "replit 2-vCPU container",
                   "wall_seconds": round(time.time() - T0, 1)},
  "honesty": "Every number above is MEASURED by this run. Fidelity = agreement%% with the kernel's ProvenanceDAG.verify_dag anomaly verdict on a held-out split. The surrogate is a fast triage; the kernel's full DAG traversal + in-toto/SLSA interop stay authoritative. Λ untouched = Conjecture 1.",
  "trained_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
with open(f"{out_dir}/TRAINING_RECEIPT.json", "w") as f:
    json.dump(receipt, f, indent=2)
print(json.dumps(receipt["metrics_MEASURED"], indent=2))
print(f"rows={len(y)} kernel_audited_dags={audit_checked} wall={receipt['environment']['wall_seconds']}s")