#!/usr/bin/env python3 """Draft 2020-12 validation plus deterministic lifecycle event replay. Expected snapshots are an oracle output, never input to the replay. Every mutable field and counter is reconstructed from ``initial`` and the ordered event log. """ from __future__ import annotations import copy import json import sys from datetime import datetime, timedelta from pathlib import Path try: from jsonschema import Draft202012Validator, FormatChecker except ImportError: print("BLOCKED: install jsonschema remotely; no fallback validation is permitted", file=sys.stderr) raise SystemExit(2) ROOT = Path(__file__).parent TERMINAL = {"complete", "failed", "cancelled", "artifact_expired"} COUNTERS = ("credit_balance", "jobs", "attempts", "debits", "compensations", "terminal_writes", "charge_cycles") errors: list[str] = [] def instant(value: str) -> datetime: return datetime.fromisoformat(value.replace("Z", "+00:00")) def close_cycle(state: dict, terminal: str) -> None: cycle = state["active_cycle"] if not cycle or cycle["closed"]: return if terminal in {"failed", "cancelled"} and cycle["funding_kind"] == "customer_debit": cycle.update(state="compensated", closed=True) state["compensations"] += 1 state["credit_balance"] += 1 else: cycle.update(state="consumed", closed=True) def terminalize(state: dict, status: str, at: str) -> None: state["status"] = status state["terminal_writes"] += 1 state["terminalized_at"] = at close_cycle(state, status) def current_lease(state: dict, event: dict) -> bool: if state.get("active_lease_token") is None: return True return (event.get("lease_token") == state.get("active_lease_token") and event.get("lease_generation") == state.get("lease_generation") and event.get("worker_id") == state.get("active_worker_id", event.get("worker_id"))) def replay(row: dict) -> dict: state = copy.deepcopy(row["initial"]) state.setdefault("active_worker_id", None) state.setdefault("source_present", True) state.setdefault("private_copy_present", False) state.setdefault("zip_present", False) state.setdefault("publication_present", False) state.setdefault("account_operation_state", None) state.setdefault("terminalized_at", None) state.setdefault("cleanup_acks", set()) state.setdefault("cancel_requests", set()) seen_idempotency: dict[tuple[str, str], str] = {} outcome, error = "ignore", None for event in row["events"]: kind, now = event["type"], instant(event["at"]) outcome, error = "ignore", None if kind == "submit": key = (event["owner_id"], event["idempotency_key"]) if key in seen_idempotency: outcome = "idempotent_replay" if seen_idempotency[key] == event["request_hash"] else "reject" error = None if outcome == "idempotent_replay" else "STATE_CONFLICT" elif not event.get("acceptance_committed", True): outcome, error = "reject", "ACCEPTANCE_ROLLBACK" else: seen_idempotency[key] = event["request_hash"] state.update(status="queued", jobs=state["jobs"] + 1, debits=state["debits"] + 1, charge_cycles=state["charge_cycles"] + 1, credit_balance=state["credit_balance"] - 1) state["active_cycle"] = {"ordinal": 1, "funding_kind": "customer_debit", "state": "open", "closed": False} state["source_present"] = event["source_present"] state["source_expires_at"] = event["source_expires_at"] outcome = "apply" elif kind == "claim": usable = event.get("private_copy_present", False) or (event.get("source_present", False) and now < instant(state["source_expires_at"])) if state["status"] != "queued": outcome, error = "reject", "STATE_CONFLICT" elif not usable: terminalize(state, "failed", event["at"]); outcome, error = "apply", "SOURCE_EXPIRED" else: state.update(status="processing", attempts=state["attempts"] + 1, active_worker_id=event["worker_id"], active_lease_token=event["lease_token"], lease_generation=event["lease_generation"], lease_expires_at=event["lease_expires_at"]) outcome = "apply" elif kind == "heartbeat": if not current_lease(state, event): outcome, error = "reject", "LEASE_FENCED" elif now > instant(state["lease_expires_at"]): outcome, error = "reject", "LEASE_FENCED" else: outcome = "apply" elif kind == "reclaim": if now <= instant(state["heartbeat_grace_until"]): outcome, error = "reject", "HEARTBEAT_GRACE_ACTIVE" elif not event["termination_confirmed"]: outcome, error = "reject", "TERMINATION_UNCONFIRMED" elif state["attempts"] >= state["attempt_budget"]: outcome, error = "reject", "ATTEMPT_BUDGET_EXHAUSTED" elif not (event.get("source_present", False) or event.get("private_copy_present", False)) or now >= instant(state["source_expires_at"]): terminalize(state, "failed", event["at"]); outcome, error = "apply", "SOURCE_EXPIRED" else: # Reclaim only releases the fenced attempt. A later claim owns the increment. state.update(status="queued", active_worker_id=None, active_lease_token=None) outcome = "apply" elif kind in {"complete", "fail"}: if not current_lease(state, event): outcome, error = "reject", "LEASE_FENCED" elif state["status"] not in {"processing", "cancellation_pending"}: outcome = "ignore" else: terminalize(state, "complete" if kind == "complete" else "failed", event["at"]) if kind == "complete": state["zip_present"] = True; state["publication_present"] = True state["completed_at"] = event["at"] state["zip_expires_at"] = event.get("zip_expires_at") outcome = "apply" elif kind == "cancel": authority = next((a for a in row["refs"]["actor_roles"] if a["actor_id"] == event["actor_id"]), None) authorized = (event["owner_id"] == row["refs"]["owner_id"] and authority is not None and authority["role"] == "owner" and kind in authority["allowed_operations"]) if not authorized: outcome, error = "reject", "NOT_FOUND" elif state["status"] == "cancelled": outcome = "idempotent_replay" elif state["status"] in TERMINAL: outcome, error = "reject", "STATE_CONFLICT" elif event["cancel_request_id"] in state["cancel_requests"]: outcome = "idempotent_replay" else: state["cancel_requests"].add(event["cancel_request_id"]) state["status"] = "cancellation_pending"; outcome = "apply" elif kind == "cleanup_ack": if event["cancel_request_id"] not in state["cancel_requests"]: outcome, error = "reject", "STATE_CONFLICT" elif event["cancel_request_id"] in state["cleanup_acks"]: outcome = "idempotent_replay" elif not current_lease(state, event): outcome, error = "reject", "LEASE_FENCED" else: state["cleanup_acks"].add(event["cancel_request_id"]) terminalize(state, "cancelled", event["at"]); outcome = "apply" elif kind == "cleanup_failed": outcome, error = "apply", "CLEANUP_FAILED" elif kind in {"terminate", "worker_crash"}: outcome = "apply" if event["termination_confirmed"] else "reject" elif kind == "admin_retry": key = (event["operator_id"], event["idempotency_key"]) if key in seen_idempotency: outcome = "idempotent_replay" elif state["status"] != "failed" or not state.get("retryable", False): outcome, error = "reject", "STATE_CONFLICT" else: seen_idempotency[key] = "retry"; state["status"] = "queued" state["attempts"] += 1; state["charge_cycles"] += 1 state["active_cycle"] = {"ordinal": state["charge_cycles"], "funding_kind": "goodwill", "state": "open", "closed": False} outcome = "apply" elif kind == "artifact_expire": if state["status"] != "complete" or now < instant(state["zip_expires_at"]): outcome, error = "reject", "STATE_CONFLICT" else: terminalize(state, "artifact_expired", event["at"]) state.update(zip_present=False, publication_present=False); outcome = "apply" elif kind == "artifact_publish": if state["status"] != "complete" or not state["zip_present"]: outcome, error = "reject", "STATE_CONFLICT" else: state["publication_present"] = True; outcome = "apply" elif kind == "private_copy_create": if not state["source_present"]: outcome, error = "reject", "SOURCE_EXPIRED" else: state["private_copy_present"] = True; outcome = "apply" elif kind == "zip_delete": if not state["zip_present"]: outcome = "ignore" else: state["zip_present"] = False; outcome = "apply" elif kind == "source_delete": expiry = instant(state["source_expires_at"]) if now < expiry or (not state["source_present"] and not state["private_copy_present"]): outcome = "ignore" else: state["source_present"] = False; outcome = "apply" elif kind == "account_operation_terminalize": if state["account_operation_state"] in {"succeeded", "failed"}: outcome = "idempotent_replay" else: state["account_operation_state"] = event["terminal_state"]; outcome = "apply" else: raise ValueError(f"unsupported event {kind}") return {"outcome": outcome, "final_status": state["status"], "counts": {k: state[k] for k in COUNTERS}, "active_cycle": state["active_cycle"], "error": error, "effects": {"source_present": state["source_present"], "private_copy_present": state["private_copy_present"], "zip_present": state["zip_present"], "publication_present": state["publication_present"], "account_operation_state": state["account_operation_state"], "terminalized_at": state["terminalized_at"]}} def semantic(row: dict) -> list[str]: out: list[str] = [] events, refs = row["events"], row["refs"] if [e["seq"] for e in events] != list(range(1, len(events) + 1)): out.append("event seq must be contiguous from 1") if [instant(e["at"]) for e in events] != sorted(instant(e["at"]) for e in events): out.append("event clocks must be monotonic") identity_refs = {"invocation_id": "invocation_ids", "attempt_id": "attempt_ids", "cycle_id": "cycle_ids", "account_id": "account_ids", "object_id": "object_ids", "ledger_id": "ledger_ids", "barrier_id": "barrier_ids"} role_for = {"submit": "owner", "cancel": "owner", "claim": "worker", "heartbeat": "worker", "complete": "worker", "fail": "worker", "cleanup_ack": "worker", "cleanup_failed": "worker", "admin_retry": "operator", "reclaim": "system", "terminate": "system", "worker_crash": "system", "artifact_publish": "system", "private_copy_create": "system", "source_delete": "system", "zip_delete": "system", "artifact_expire": "system", "account_operation_terminalize": "system"} authorities = {(a["actor_id"], a["role"]): a for a in refs["actor_roles"]} for event in events: authority = authorities.get((event["actor_id"], role_for[event["type"]])) authoritative_owner = event["owner_id"] if authority and authority["role"] == "owner" else refs["owner_id"] if authority is None or event["type"] not in authority["allowed_operations"] or authority["owner_id"] != authoritative_owner: out.append(f"actor {event['actor_id']} lacks authoritative role/operation for {event['type']}") if event["conversion_id"] != refs["conversion_id"] or event["job_id"] != refs["job_id"]: out.append("foreign conversion/job reference") for field, collection in identity_refs.items(): if event[field] not in refs[collection]: out.append(f"foreign {field} reference") if "owner_id" in event and event["owner_id"] not in refs.get("owner_ids", [refs["owner_id"]]): out.append("foreign owner reference") if "worker_id" in event and event["worker_id"] not in refs["worker_ids"]: out.append("foreign worker reference") if "lease_token" in event and event["lease_token"] not in refs["lease_tokens"]: out.append("foreign lease token reference") if "lease_generation" in event and event["lease_generation"] not in refs["lease_generations"]: out.append("foreign lease generation reference") if "cancel_request_id" in event and event["cancel_request_id"] not in refs["cancel_request_ids"]: out.append("foreign cancellation request reference") auth = row["authoritative"] derived_source_expiry = instant(auth["source_accepted_at"]) + timedelta(seconds=auth["source_ttl_seconds"]) initial_source_expiry = row["initial"].get("source_expires_at") if initial_source_expiry and instant(initial_source_expiry) != derived_source_expiry: out.append("initial source clock is not config/event-derived") if "lease_expires_at" in row["initial"]: if "lease_acquired_at" not in auth or instant(row["initial"]["lease_expires_at"]) != instant(auth["lease_acquired_at"]) + timedelta(seconds=auth["lease_ttl_seconds"]): out.append("initial lease clock is not config/event-derived") if instant(row["initial"]["heartbeat_grace_until"]) != instant(row["initial"]["lease_expires_at"]) + timedelta(seconds=auth["heartbeat_grace_seconds"]): out.append("heartbeat grace clock is not config-derived") for event in (e for e in events if e["type"] == "submit"): if instant(event["at"]) != instant(event["accepted_at"]): out.append("accepted_at must be authoritative event clock") if instant(event["source_expires_at"]) != instant(event["accepted_at"]) + timedelta(hours=48): out.append("source expiry must equal accepted_at + 48h") for event in (e for e in events if e["type"] == "complete" and e.get("zip_expires_at")): if instant(event["zip_expires_at"]) != instant(event["at"]) + timedelta(hours=48): out.append("ZIP expiry must equal completed_at + 48h") concurrency = row.get("concurrency") if concurrency: actors = {e["actor_id"] for e in events} attempts = concurrency["unordered_attempts"] if len(actors) < 2 or len(attempts) < 2: out.append("concurrency requires two real event actors and attempts") by_seq = {e["seq"]: e for e in events} for attempt in attempts: linked = by_seq.get(attempt["event_seq"]) if (attempt["actor_id"] not in concurrency["actors"] or attempt["barrier_id"] != concurrency["barrier_id"] or linked is None or any(attempt[k] != linked[k] for k in ("attempt_id", "actor_id")) or attempt["event_type"] != linked["type"]): out.append("unlinked concurrency attempt") for actor in concurrency["actors"]: if actor not in {a["actor_id"] for a in attempts}: out.append("concurrency actor has no attempt") contenders = [by_seq[a["event_seq"]] for a in attempts if a["event_seq"] in by_seq] ordered = sorted(contenders, key=lambda e: (instant(e["at"]), e["seq"])) winners = [e for e in contenders if e["winner"]] if len(winners) != 1 or not ordered or winners[0]["seq"] != ordered[0]["seq"]: out.append("winner must derive from event/barrier ordering") declared = {(w["event_type"], w["final_status"]) for w in concurrency["allowed_winners"]} produced = set() contender_seqs = {a["event_seq"] for a in attempts} for attempt in attempts: permutation = copy.deepcopy(row) permutation.pop("concurrency", None) chosen = next(e for e in permutation["events"] if e["seq"] == attempt["event_seq"]) permutation["events"] = [e for e in permutation["events"] if e["seq"] not in contender_seqs or e["seq"] == attempt["event_seq"]] if "worker_id" in chosen and chosen["type"] in {"complete", "fail", "cleanup_ack"}: permutation["initial"]["active_worker_id"] = chosen["worker_id"] result = replay(permutation) produced.add((chosen["type"], result["final_status"])) if declared != produced: out.append(f"allowed_winners must exactly match executed permutations: declared {sorted(declared)!r}, derived {sorted(produced)!r}") actual = replay(row) expected = {k: row["expected"][k] for k in ("outcome", "final_status", "counts", "active_cycle", "error", "effects")} operation_effects = {"complete": "zip_present", "fail": "terminalized_at", "cleanup_ack": "terminalized_at", "artifact_expire": "publication_present", "source_delete": "source_present", "account_operation_terminalize": "account_operation_state", "artifact_publish": "publication_present", "private_copy_create": "private_copy_present", "zip_delete": "zip_present"} event_types = {event["type"] for event in events} for operation, effect_field in operation_effects.items(): if operation in event_types and expected["effects"].get(effect_field) != actual["effects"].get(effect_field): out.append(f"terminalizer {operation} effect {effect_field} mismatch") if actual != expected: out.append(f"replay mismatch: expected {expected!r}, derived {actual!r}") if row["case_id"] == "K9": matrix = [] for item in row["terminal_matrix"]: authoritative_snapshot = copy.deepcopy(row["initial"]) authoritative_snapshot["status"] = item["initial_status"] if item["expected_snapshot"] != authoritative_snapshot: out.append(f"K9 terminal snapshot mutated for {item['initial_status']}") probe = copy.deepcopy(row) probe["initial"] = copy.deepcopy(item["expected_snapshot"]) result = replay(probe) expected_projection = {"final_status": item["initial_status"], "counts": {k: item["expected_snapshot"][k] for k in COUNTERS}, "active_cycle": item["expected_snapshot"]["active_cycle"], "effects": {k: item["expected_snapshot"].get(k, default) for k, default in (("source_present", True), ("private_copy_present", False), ("zip_present", False), ("publication_present", False), ("account_operation_state", None), ("terminalized_at", None))}} cancel_result = "idempotent" if result["outcome"] == "idempotent_replay" else "state_conflict" if cancel_result != item["cancel_result"] or any(result[k] != expected_projection[k] for k in expected_projection): out.append(f"K9 terminal snapshot mutated for {item['initial_status']}") return out pairs = (("lifecycle-case.schema.json", "lifecycle-cases.jsonl"), ("public-error-case.schema.json", "public-error-cases.jsonl"), ("blocked-conformance-case.schema.json", "blocked-conformance-cases.jsonl")) records, validators = {}, {} for schema_name, data_name in pairs: schema = json.loads((ROOT / schema_name).read_text()); Draft202012Validator.check_schema(schema) validator = Draft202012Validator(schema, format_checker=FormatChecker()); validators[data_name] = validator rows = [] for line_number, line in enumerate((ROOT / data_name).read_text().splitlines(), 1): if not line.strip(): continue row = json.loads(line); rows.append(row) for exc in validator.iter_errors(row): errors.append(f"{data_name}:{line_number}:{'.'.join(map(str, exc.absolute_path)) or '$'}: {exc.message}") records[data_name] = rows life = records["lifecycle-cases.jsonl"] for data_name, rows in records.items(): seen_case_ids = set() for row in rows: case_id = row["case_id"] if case_id in seen_case_ids: errors.append(f"{data_name}: duplicate case_id {case_id}") seen_case_ids.add(case_id) if "--regenerate" in sys.argv: for row in life: evidence = row["expected"].get("evidence", ["derived by ordered event replay oracle"]) row["expected"] = {**replay(row), "evidence": evidence} (ROOT / "lifecycle-cases.jsonl").write_text("\n".join(json.dumps(row, separators=(",", ":")) for row in life) + "\n") print(f"REGENERATED: {len(life)} expected snapshots from ordered replay") raise SystemExit(0) required = {*(f"K{i}" for i in range(1, 11)), *(f"CANCEL-{i}" for i in range(1, 8))} if required - {r["case_id"] for r in life}: errors.append(f"missing canonical cases: {sorted(required - {r['case_id'] for r in life})}") for row in life: try: errors.extend(f"{row['case_id']}: {message}" for message in semantic(row)) except Exception as exc: errors.append(f"{row.get('case_id', '?')}: replay crashed: {exc}") # Mutation-complete negative suite: each family must be independently observed. def lifecycle_rejections(row): schema_errors = [f"schema:{exc.message}" for exc in validators["lifecycle-cases.jsonl"].iter_errors(row)] if schema_errors: return schema_errors return semantic(row) def must_fail(name, row, mutate, required_rejection=None): probe = copy.deepcopy(row); mutate(probe) rejections = lifecycle_rejections(probe) if not rejections: errors.append(f"SELFTEST {name}: invalid fixture accepted") elif required_rejection and not any(required_rejection in rejection for rejection in rejections): errors.append(f"SELFTEST {name}: expected {required_rejection!r}, got {rejections!r}") if life: base = next(r for r in life if r["case_id"] == "JL-001") mutations = { "transition": lambda r: r["initial"].__setitem__("status", "complete"), "discriminator": lambda r: r["events"][0].__setitem__("type", "unknown_terminalizer"), "format": lambda r: r["events"][0].__setitem__("at", "not-an-instant"), "min-generation": lambda r: r["events"][0].__setitem__("lease_generation", -1), "link": lambda r: r["events"][0].__setitem__("job_id", "job_FOREIGN"), "count": lambda r: r["expected"]["counts"].__setitem__("attempts", 99), "clock": lambda r: r["authoritative"].__setitem__("source_accepted_at", "2026-08-21T12:59:59.998Z"), "boundary": lambda r: r["initial"].__setitem__("source_expires_at", "2026-08-23T12:59:59.999Z"), "identity": lambda r: r["events"][0].__setitem__("attempt_id", "attempt_FOREIGN"), "actor": lambda r: r["refs"]["actor_roles"][0]["allowed_operations"].clear(), "ref": lambda r: r["refs"].__setitem__("conversion_id", "conversion_FOREIGN"), } for name, mutate in mutations.items(): must_fail(name, base, mutate) race = next(r for r in life if r.get("concurrency")) must_fail("unordered-actor", race, lambda r: r["concurrency"]["actors"].pop()) must_fail("unordered-attempt", race, lambda r: r["concurrency"]["unordered_attempts"].pop()) must_fail("winner", race, lambda r: [e.__setitem__("winner", False) for e in r["events"]]) operation_effects = { "complete": "zip_present", "fail": "terminalized_at", "cleanup_ack": "terminalized_at", "artifact_expire": "publication_present", "source_delete": "source_present", "account_operation_terminalize": "account_operation_state", "artifact_publish": "publication_present", "private_copy_create": "private_copy_present", "zip_delete": "zip_present", } for family, effect_field in operation_effects.items(): fixture = next(r for r in life if any(e["type"] == family for e in r["events"])) def mutate_effect(r, field=effect_field): current = r["expected"]["effects"][field] if isinstance(current, bool): r["expected"]["effects"][field] = not current elif field == "terminalized_at": r["expected"]["effects"][field] = "2026-08-22T00:00:00.000Z" elif field == "account_operation_state": r["expected"]["effects"][field] = "failed" if current != "failed" else "succeeded" else: raise AssertionError(field) must_fail(f"terminalizer-{family}", fixture, mutate_effect, f"terminalizer {family} effect {effect_field} mismatch") k9 = next(r for r in life if r["case_id"] == "K9") for field in ("status", "active_cycle", "zip_present", "source_present", "account_operation_state"): def mutate_snapshot(r, field=field): snapshot = r["terminal_matrix"][0]["expected_snapshot"] defaults = {"zip_present": False, "source_present": True, "account_operation_state": None} current = snapshot.get(field, defaults.get(field)) if field in {"zip_present", "source_present"}: snapshot[field] = not current elif field == "status": snapshot[field] = "failed" elif field == "active_cycle": snapshot[field] = {**current, "state": "open", "closed": False} elif field == "account_operation_state": snapshot[field] = "failed" else: raise AssertionError(field) must_fail(f"K9-snapshot-{field}", k9, mutate_snapshot, "K9 terminal snapshot mutated for complete") # The public contract is also exercised negatively through its full Draft 2020-12 validator. for index, public_row in enumerate(records["public-error-cases.jsonl"], 1): probe = copy.deepcopy(public_row) required_field = next(iter(json.loads((ROOT / "public-error-case.schema.json").read_text())["required"])) probe.pop(required_field, None) if not list(validators["public-error-cases.jsonl"].iter_errors(probe)): errors.append(f"SELFTEST public-{index}: malformed public error accepted") if errors: print("FAIL: fixture validation", file=sys.stderr) print("\n".join(f"- {error}" for error in errors), file=sys.stderr) raise SystemExit(1) print(f"PASS: Draft 2020-12 + formats + deterministic event replay + mutation self-tests ({len(life)} lifecycle cases)")