#!/usr/bin/env python3 """Original stdlib trace illustration. Declared evidence is never authenticated here.""" import argparse import copy import json import re from collections import Counter from pathlib import Path SCHEMA = "nodus.startup-milestone-demo/1" MODES = {"fresh_instance", "warm_host_process", "retained_state"} STATES = {"accepted", "failed", "validation_failed", "timed_out", "unavailable", "not_started", "unresolved"} STAGES = ("submit", "ready_instance", "first_cuda", "validation_observed", "durable_read_observed", "observation_end") def require(ok, code): if not ok: raise ValueError(code) def integer(value, code, minimum=0): require(type(value) is int and minimum <= value <= 9_000_000_000_000_000, code) def digest(value): return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None def elapsed(start, end, clocks): """Return a declared [lower, upper] interval, or unknown. Never clip negatives.""" a, b = clocks[start["clock"]], clocks[end["clock"]] difference = end["t_ms"] - start["t_ms"] if start["clock"] == end["clock"] and a["kind"] == "monotonic": error = 2 * a["error_bound_ms"] return {"bounds_ms": [difference - error, difference + error], "basis": "same_declared_monotonic_clock_instance"} if a["kind"] == b["kind"] == "utc": am, bm = a.get("mapping"), b.get("mapping") if am and bm and am["domain"] == bm["domain"]: error = am["error_bound_ms"] + bm["error_bound_ms"] return {"bounds_ms": [difference - error, difference + error], "basis": "declared_common_UTC_mapping_and_error_bounds"} return {"bounds_ms": None, "basis": "unknown_cross_clock_or_unbounded_wall_clock"} def report(trace): require(trace.get("schema") == SCHEMA, "schema") require(trace.get("evidence_kind") in {"synthetic_fixture", "declared_observation"}, "unfilled_or_unknown_evidence_kind") clocks = trace.get("clocks", {}) require(isinstance(clocks, dict) and clocks, "clocks") for cid, clock in clocks.items(): require(isinstance(cid, str) and cid and isinstance(clock, dict), "clock_id") require(clock.get("kind") in {"monotonic", "utc"}, "clock_kind") require(clock.get("observer") in {"caller", "worker"}, "clock_observer") require(isinstance(clock.get("epoch"), str) and clock["epoch"], "clock_epoch") if clock["kind"] == "monotonic": integer(clock.get("error_bound_ms"), "monotonic_error_bound") if clock.get("mapping") is not None: m = clock["mapping"] require(clock["kind"] == "utc" and isinstance(m, dict), "mapping_kind") require(all(isinstance(m.get(k), str) and m[k] for k in ["domain", "evidence_ref"]), "mapping_provenance") for key in ["valid_from_ms", "valid_until_ms", "error_bound_ms"]: integer(m.get(key), "mapping_bounds") require(m["valid_from_ms"] <= m["valid_until_ms"], "mapping_window") strata = trace.get("strata", {}) require(isinstance(strata, dict) and strata, "strata") for row in strata.values(): require(row.get("mode") in MODES, "coldness_mode") require(all(isinstance(row.get(k), str) and row[k] for k in ["cache_state", "preparation_scope", "workload_ref"]), "stratum_scope") plan, attempts = trace.get("plan"), trace.get("attempts") require(isinstance(plan, list) and plan and isinstance(attempts, list), "plan") ids = [p.get("attempt_id") for p in plan] require(all(isinstance(i, str) and i for i in ids) and len(set(ids)) == len(ids), "unique_plan_ids") require([a.get("attempt_id") for a in attempts] == ids, "complete_ordered_plan") rows = [] for p, attempt in zip(plan, attempts): require(p.get("stratum_id") in strata, "planned_stratum") integer(p.get("deadline_ms"), "deadline", 1) require(attempt.get("status") in STATES, "status") events = attempt.get("events") require(isinstance(events, dict) and set(events) <= set(STAGES), "events") for stage, event in events.items(): require(isinstance(event, dict) and event.get("clock") in clocks, "event_clock") integer(event.get("t_ms"), "event_time") c = clocks[event["clock"]] if c.get("mapping"): m = c["mapping"] require(m["valid_from_ms"] <= event["t_ms"] <= m["valid_until_ms"], "event_outside_mapping_window") if attempt["status"] == "not_started": require(not events, "not_started_has_events") else: require("submit" in events and "observation_end" in events, "observation_window") require(clocks[events["submit"]["clock"]]["observer"] == "caller", "caller_submit") present = [s for s in STAGES if s in events] for i, before in enumerate(present): for after in present[i + 1:]: interval = elapsed(events[before], events[after], clocks)["bounds_ms"] require(interval is None or interval[1] >= 0, "contradictory_event_order") validation, read = events.get("validation_observed"), events.get("durable_read_observed") if validation: require(type(validation.get("accepted")) is bool and digest(validation.get("artifact_sha256")) and isinstance(validation.get("evidence_id"), str) and validation["evidence_id"], "validation_binding") if read: require(digest(read.get("artifact_sha256")) and read.get("read_acknowledged") is True, "durable_read_binding") require(clocks[read["clock"]]["observer"] == "caller", "caller_read_ack") accepted = attempt["status"] == "accepted" if accepted: require(validation and validation["accepted"] and read, "accepted_needs_validation_and_read") require(validation["artifact_sha256"] == read["artifact_sha256"] and read.get("after_validation_id") == validation["evidence_id"], "acceptance_identity_and_order") elif validation and validation["accepted"] and read: require(read.get("after_validation_id") != validation["evidence_id"], "accepted_outcome_not_recorded") if attempt["status"] == "validation_failed": require(validation and not validation["accepted"], "validation_failure_evidence") durations = {s: elapsed(events["submit"], events[s], clocks) for s in present if s not in {"submit", "observation_end"}} followup = elapsed(events["submit"], events["observation_end"], clocks) if events else None endpoint = durations.get("durable_read_observed") if accepted else None bounds = endpoint["bounds_ms"] if endpoint else None if accepted: deadline = ("unknown_clock" if bounds is None else "unknown_order" if bounds[0] < 0 else "within" if bounds[1] <= p["deadline_ms"] else "late" if bounds[0] > p["deadline_ms"] else "unknown_boundary_overlap") else: fb = followup["bounds_ms"] if followup else None deadline = ("not_started" if not events else "terminal_without_accepted_output" if attempt["status"] in {"failed", "unavailable", "validation_failed"} else "unknown_followup_clock" if fb is None else "right_censored_at_or_after_cutoff" if fb[0] >= p["deadline_ms"] else "unknown_cutoff_overlap" if fb[1] >= p["deadline_ms"] else "pending_or_censored_before_cutoff") if attempt["status"] == "timed_out": require(followup and followup["bounds_ms"] and followup["bounds_ms"][0] >= p["deadline_ms"], "timeout_before_proven_cutoff") rows.append({"attempt_id": p["attempt_id"], "stratum_id": p["stratum_id"], "status": attempt["status"], "milestones_from_submit": durations, "accepted_read_endpoint": endpoint, "deadline_outcome": deadline, "observed_followup": followup}) return {"schema": SCHEMA + "/report", "evidence_kind": trace["evidence_kind"], "all_numbers_fictional": trace["evidence_kind"] == "synthetic_fixture", "conformance_only": True, "evidence_authenticated": False, "planned_attempts": len(plan), "attempts": rows, "by_stratum": {sid: {"mode": s["mode"], "planned": sum(r["stratum_id"] == sid for r in rows), "status_counts": dict(Counter(r["status"] for r in rows if r["stratum_id"] == sid)), "deadline_counts": dict(Counter(r["deadline_outcome"] for r in rows if r["stratum_id"] == sid))} for sid, s in strata.items()}, "costs": {"status": "unknown", "total": None}, "limits": ["Declared refs/digests/clock bounds are not authenticated evidence.", "Read acknowledgment after acceptance is the proposed endpoint, not a universal cold-start definition.", "Durability retention scope is external; one read does not prove permanent survival.", "Do not sum overlapping stage intervals or pool coldness strata.", "No provider performance, p95, availability, idle capacity or economic claim."]} def controls(fixture): mutations = { "drop_failed_attempt": lambda t: t["attempts"].pop(4), "duplicate_plan_id": lambda t: t["plan"][1].update(attempt_id=t["plan"][0]["attempt_id"]), "ready_only_accepted": lambda t: t["attempts"][0]["events"].pop("validation_observed"), "different_artifact": lambda t: t["attempts"][0]["events"]["durable_read_observed"].update(artifact_sha256="f" * 64), "wrong_validation_reference": lambda t: t["attempts"][0]["events"]["durable_read_observed"].update(after_validation_id="other"), "read_before_validation": lambda t: t["attempts"][0]["events"]["durable_read_observed"].update(t_ms=6000), "worker_origin": lambda t: t["attempts"][0]["events"]["submit"].update(clock="worker"), "unbounded_mapping": lambda t: t["clocks"]["utc_worker"]["mapping"].update(error_bound_ms=None), "expired_mapping": lambda t: t["clocks"]["utc_worker"]["mapping"].update(valid_until_ms=1000300), "premature_timeout": lambda t: t["attempts"][-1]["events"]["observation_end"].update(t_ms=101000), } expected = ["complete_ordered_plan", "unique_plan_ids", "accepted_needs_validation_and_read", "acceptance_identity_and_order", "acceptance_identity_and_order", "contradictory_event_order", "caller_submit", "mapping_bounds", "event_outside_mapping_window", "timeout_before_proven_cutoff"] result = [] for (name, change), reason in zip(mutations.items(), expected): trace = copy.deepcopy(fixture); change(trace) try: report(trace) observed = "incorrectly_accepted" except ValueError as exc: observed = str(exc) result.append({"case": name, "expected": reason, "observed": observed, "matched": observed == reason}) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--trace", required=True, type=Path) parser.add_argument("--out", required=True, type=Path) parser.add_argument("--controls", action="store_true", help="Authored mutations of the shipped fixture only") args = parser.parse_args() trace = json.loads(args.trace.read_text(encoding="utf-8")) result = report(trace) if args.controls: require(trace["evidence_kind"] == "synthetic_fixture", "controls_are_synthetic_only") result["authored_negative_controls"] = controls(trace) text = json.dumps(result, indent=2, allow_nan=False) + "\n" with args.out.open("x", encoding="utf-8") as f: f.write(text) print(text, end="") return 0 if all(c["matched"] for c in result.get("authored_negative_controls", [])) else 1 if __name__ == "__main__": raise SystemExit(main())