"""Deterministic offline projection of the paper's synthetic example only.""" from __future__ import annotations import argparse import hashlib import json import math from pathlib import Path import sys VERSION = "synthetic-queue-projection-1" KINDS = {"ARTIFACT", "LOCAL_CHECK", "CI_CHECK", "BLOCKED", "STATUS", "REVIEW_STATUS", "HEARTBEAT"} CHECKS = {"LOCAL_CHECK", "CI_CHECK"} def number(value, label): if type(value) not in (int, float) or not math.isfinite(value) or value < 0: raise ValueError(f"{label} must be finite and nonnegative") return value def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError(f"duplicate JSON key: {key}") result[key] = value return result def load_fixture(path): raw = Path(path).read_bytes() def invalid_constant(value): raise ValueError(f"non-finite JSON number: {value}") return json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object, parse_constant=invalid_constant), hashlib.sha256(raw).hexdigest() def validate(fixture): if not isinstance(fixture, dict) or fixture.get("fixture_kind") != "synthetic_design_example_not_operational_data": raise ValueError("this checker accepts explicitly synthetic fixtures only") if fixture.get("schema_version") != "research-example-1": raise ValueError("unsupported fixture schema") if not isinstance(fixture.get("work_id"), str) or not fixture["work_id"]: raise ValueError("work_id must be a nonempty string") if not isinstance(fixture.get("events"), list): raise ValueError("events must be a list") seen, revisions, events, duplicates = {}, set(), [], [] last_time = 0 for event in fixture["events"]: if not isinstance(event, dict): raise ValueError("event must be an object") eid = event.get("event_id") if not isinstance(eid, str) or not eid: raise ValueError("event_id must be a nonempty string") if eid in seen: if json.dumps(event, sort_keys=True, allow_nan=False) != json.dumps(seen[eid], sort_keys=True, allow_nan=False): raise ValueError(f"conflicting duplicate ID: {eid}") duplicates.append(eid) continue sequence = event.get("sequence") if type(sequence) is not int or sequence != len(events) + 1: raise ValueError("unique events must have consecutive, ordered sequences starting at one") elapsed = number(event.get("elapsed_minutes"), "elapsed_minutes") if elapsed < last_time: raise ValueError("recorded elapsed time decreased") occurred = number(event.get("occurred_elapsed_minutes", elapsed), "occurred_elapsed_minutes") if occurred > elapsed: raise ValueError("occurred time exceeds recorded time in this same-clock fixture") kind, revision = event.get("type"), event.get("revision") if not isinstance(kind, str) or kind not in KINDS: raise ValueError("unsupported event type") if not isinstance(revision, str) or not revision: raise ValueError("revision must be a nonempty string") if kind == "ARTIFACT": if revision in revisions: raise ValueError("ARTIFACT must introduce a new revision") revisions.add(revision) elif revision not in revisions: raise ValueError("event references a revision not yet introduced") if kind in CHECKS and event.get("result") not in ("pass", "fail"): raise ValueError("check result must be pass or fail") causes = event.get("causes", []) if not isinstance(causes, list) or any(not isinstance(c, str) for c in causes): raise ValueError("causes must be a list of event IDs") if len(set(causes)) != len(causes) or any(c not in seen for c in causes): raise ValueError("causes must identify distinct earlier events") if any(seen[c].get("occurred_elapsed_minutes", seen[c]["elapsed_minutes"]) > occurred for c in causes): raise ValueError("a same-clock cause cannot occur after its child event") seen[eid] = event events.append(event) last_time = elapsed return events, duplicates def project(fixture, cutoff_sequence=None, cutoff_minutes=None): events, duplicates = validate(fixture) cutoff = fixture.get("cutoff_sequence") if cutoff_sequence is None else cutoff_sequence if type(cutoff) is not int or not 0 <= cutoff <= len(events): raise ValueError("cutoff_sequence must select an existing prefix, including zero") prefix = events[:cutoff] last_time = prefix[-1]["elapsed_minutes"] if prefix else 0 clock = last_time if cutoff_minutes is None else number(cutoff_minutes, "cutoff_minutes") if clock < last_time or (cutoff < len(events) and clock > events[cutoff]["elapsed_minutes"]): raise ValueError("cutoff time must follow included events and not exceed the next event time") current = current_artifact = last_progress = None receipts, ci, episodes, repeated, active = [], {}, [], [], None for event in prefix: kind, revision, eid = event["type"], event["revision"], event["event_id"] if kind == "ARTIFACT": current, current_artifact = revision, eid last_progress = event if active and set(event.get("causes", [])) & set(active["blocker_event_ids"]): active.update(resumption_event_id=eid, ended_at_minutes=event["elapsed_minutes"], duration_minutes=event["elapsed_minutes"] - active["started_at_minutes"], right_censored=False) active = None elif kind in CHECKS: receipts.append({"event_id": eid, "revision": revision, "type": kind, "result": event["result"], "stale_on_arrival": revision != current}) if kind == "CI_CHECK": ci[revision] = {"event_id": eid, "result": event["result"]} if revision == current: last_progress = event elif kind == "BLOCKED": if revision != current: raise ValueError("BLOCKED must refer to the current revision") if active: if not set(event.get("causes", [])) & set(active["blocker_event_ids"]): raise ValueError("simultaneous unrelated blockers are unsupported") active["blocker_event_ids"].append(eid) repeated.append(eid) else: active = {"start_event_id": eid, "blocker_event_ids": [eid], "started_at_minutes": event["elapsed_minutes"], "resumption_event_id": None, "ended_at_minutes": None, "duration_minutes": None, "right_censored": True} episodes.append(active) if active: active["duration_minutes"] = clock - active["started_at_minutes"] latest_ci = next((r for r in reversed(receipts) if r["type"] == "CI_CHECK"), None) return {"projection_version": VERSION, "fixture_kind": fixture["fixture_kind"], "work_id": fixture["work_id"], "cutoff_sequence": cutoff, "cutoff_minutes": clock, "unique_events_at_cutoff": len(prefix), "input_duplicate_event_ids": duplicates, "current_revision": current, "current_artifact_event_id": current_artifact, "current_revision_reported_ci": ci.get(current), "last_arriving_ci": latest_ci, "last_arriving_ci_matches_current_revision": bool(latest_ci and latest_ci["revision"] == current), "superseded_receipt_ids": [r["event_id"] for r in receipts if r["revision"] != current], "stale_on_arrival_receipt_ids": [r["event_id"] for r in receipts if r["stale_on_arrival"]], "repeated_blocker_event_ids": repeated, "last_message": {"event_id": prefix[-1]["event_id"], "elapsed_minutes": last_time} if prefix else None, "last_evidenced_progress": {"event_id": last_progress["event_id"], "elapsed_minutes": last_progress["elapsed_minutes"]} if last_progress else None, "blocker_episodes": episodes, "permission": "not_established_by_this_fixture"} def check_expected(fixture, result): """Compare the six authored full-history answers only after projection.""" episodes = result["blocker_episodes"] ci = result["current_revision_reported_ci"] duration = episodes[0]["duration_minutes"] if episodes and not episodes[0]["right_censored"] else None answers = { "Which is the current artifact revision?": result["current_revision"], "Can the last-arriving CI result certify r2?": result["last_arriving_ci_matches_current_revision"], "Did the repeated blocker at minute 8 demonstrate progress?": False if result["repeated_blocker_event_ids"] else None, "What is the synthetic blocker-to-resumption duration?": duration, "Do these events establish permission to merge?": False, "What supports the r2 CI status?": f"{ci['event_id']} {ci['result']}" if ci else None, } expected = fixture.get("expected_answers") if not isinstance(expected, list) or len(expected) != len(answers): raise ValueError("expected oracle must contain the six authored questions") questions = [item.get("question") for item in expected if isinstance(item, dict)] if len(questions) != len(answers) or set(questions) != set(answers): raise ValueError("expected oracle has missing, duplicate or unsupported questions") failed = [item["question"] for item in expected if answers[item["question"]] != item.get("answer", item.get("answer_minutes"))] return {"questions_checked": len(expected), "failed_questions": failed, "all_match": not failed} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("fixture", nargs="?", type=Path, default=Path(__file__).with_name("replay_fixture.json")) parser.add_argument("--cutoff-sequence", type=int) parser.add_argument("--cutoff-minutes", type=float) parser.add_argument("--check-expected", action="store_true") args = parser.parse_args(argv) try: if args.check_expected and (args.cutoff_sequence is not None or args.cutoff_minutes is not None): raise ValueError("expected answers apply only to the full authored cutoff") fixture, digest = load_fixture(args.fixture) result = project(fixture, args.cutoff_sequence, args.cutoff_minutes) result["input_sha256"] = digest if args.check_expected: result["expected_answer_check"] = check_expected(fixture, result) print(json.dumps(result, indent=2, sort_keys=True, allow_nan=False)) return 0 if result.get("expected_answer_check", {}).get("all_match", True) else 1 except (ValueError, OSError) as error: parser.exit(2, f"error: {error}\n") if __name__ == "__main__": sys.exit(main())