net-kingdom/tools/execution-attribution-receipt/generate_fixtures.py

#!/usr/bin/env python3
"""Regenerate the valid fixtures from the real executor (bounded-executor/executor.py) on in-memory fakes.

Each scenario runs executor.Run.execute() on the pilot example admission with fixed clocks and a fixed run id, and
writes the events it emitted to fixtures/valid/<scenario>-{started,receipt}.json. The fixtures therefore always
have the executor's exact shape; tests fail when the executor and the schema drift apart. No value an operation
handles can reach them: the fake operation holds a marker value, and the generator refuses if it appears.

  python3 tools/execution-attribution-receipt/generate_fixtures.py [--check]
"""
from __future__ import annotations

import argparse
import copy
import datetime as dt
import importlib.util
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).resolve().parent
ROOT = HERE.parents[1]
OUT = HERE / "fixtures/valid"
MARKER = "synthetic-value-never-emitted"
APPROVAL = "6f1c2b9e-0a4d-4e3b-9c51-7d2f8a6b3e10"
NOW = dt.datetime(2026, 11, 3, 9, 30, tzinfo=dt.UTC)


def _executor():
    spec = importlib.util.spec_from_file_location("executor", ROOT / "bounded-executor/executor.py")
    module = importlib.util.module_from_spec(spec)
    sys.modules["executor"] = module
    spec.loader.exec_module(module)
    return module


X = _executor()


class Fake:
    def __init__(self, doc, fail_checks=()):
        spec = doc["spec"]
        self.t = 0.0
        self.fail = list(fail_checks)
        self.events = []
        self.approval = {
            "id": APPROVAL, "status": "approved",
            "binding": {"action": X.ACTION, "actor": spec["executor"]["identity"], "principal": "admin",
                        "purpose": spec["purpose"], "human_control": True, "pdp_path": False,
                        "target": {"admission_id": spec["id"], "admission_digest": X.bea.admission_digest(doc)}},
            "validity": {"not_before": spec["window"]["not_before"], "expires_at": spec["window"]["not_after"]},
            "entries": [{"subject_id": "kc-sub-0f3a", "principal_type": "human", "approved_at": "2026-11-03T09:10:00Z",
                         "assurance": json.dumps({"level": "aal2", "mfa": True, "source": "key-cape"})}]}

    def get(self, _id):
        return copy.deepcopy(self.approval)

    def consume(self, _id, digest, decision_id):
        self.approval.update(status="consumed", consumption={"request_digest": digest, "decision_id": decision_id,
                                                             "consumed_at": "2026-11-03T09:30:00Z"})
        return {"status": "consumed", "request_digest": digest, "decision_id": decision_id,
                "consumed_at": "2026-11-03T09:30:00Z", "idempotent": False}

    def op(self, _step, ctx):
        ctx.data["value"] = MARKER
        self.t += 5

    def check(self, check, _ctx):
        self.t += 2
        if self.fail and self.fail[0] == check["id"]:
            self.fail.pop(0)
            return False
        return True

    def sleep(self, seconds):
        self.t += seconds


def scenario(name: str, fail_checks=(), attempts=1) -> list[dict]:
    doc = X.bea.load(ROOT / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml")
    doc["spec"]["budget"]["attempts"] = attempts
    doc["spec"]["plan_digest"] = X.bea.plan_digest(doc)
    fake = Fake(doc, fail_checks)
    plan = doc["spec"]["plan"]
    deps = X.Deps(carrier=fake, operations={s["operation"]: fake.op for s in plan["steps"] + plan["revert_steps"]},
                  checks={c["id"]: fake.check for c in doc["spec"]["checks"]}, trust=lambda _s: True,
                  audit=type("Sink", (), {"emit": lambda _s, e: fake.events.append(e)})(), alert=lambda *_a: None,
                  wall=lambda: NOW + dt.timedelta(seconds=fake.t), monotonic=lambda: fake.t, sleep=fake.sleep)
    cfg = X.Config(executor_identity=doc["spec"]["executor"]["identity"],
                   runtime_principal="system:serviceaccount:sso:nk-bind-executor", tenant="platform",
                   environment="PRODUCTION", approver_subjects={"admin": "kc-sub-0f3a"},
                   approver_issuer="https://kc.coulomb.social")
    run = X.Run(doc, APPROVAL, cfg, deps)
    run.run_id = {"completed": "0b7e4a52-3c1d-4f6e-8a90-1b2c3d4e5f60",
                  "reverted": "1c8f5b63-4d2e-4a7f-9ba1-2c3d4e5f6071",
                  "completed-on-retry": "2d906c74-5e3f-4b80-8cb2-3d4e5f607182"}[name]
    run.preflight = lambda: run._check_approval(fake.get(APPROVAL), consumed_ok=False)  # pins: example, not repo
    result = run.execute()
    assert result["outcome"] in ("completed", "reverted"), result
    text = json.dumps(fake.events)
    if MARKER in text:
        raise SystemExit("an operation value reached an event")
    return fake.events


SCENARIOS = {
    "completed": {},
    "reverted": {"fail_checks": ["synthetic-journey"]},
    "completed-on-retry": {"fail_checks": ["synthetic-journey"], "attempts": 2},
}


def render() -> dict[str, str]:
    files = {}
    for name, kwargs in SCENARIOS.items():
        started, receipt = scenario(name, **kwargs)
        for kind, event in (("started", started), ("receipt", receipt)):
            files[f"{name}-{kind}.json"] = json.dumps(event, indent=2, sort_keys=True) + "\n"
    return files


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
    parser.add_argument("--check", action="store_true", help="fail if the committed fixtures differ")
    args = parser.parse_args(argv)
    files = render()
    if args.check:
        stale = [n for n, text in files.items() if not (OUT / n).is_file() or (OUT / n).read_text() != text]
        print(json.dumps({"stale": stale}))
        return 1 if stale else 0
    OUT.mkdir(parents=True, exist_ok=True)
    for name, text in files.items():
        (OUT / name).write_text(text)
    print(f"wrote {len(files)} fixtures to {OUT.relative_to(ROOT)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())