Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229#!/usr/bin/env python3"""Validate portable state for an event-sourced garbage-collection run."""from __future__ import annotations
import argparseimport jsonimport reimport sysimport tempfilefrom pathlib import Pathfrom typing import Any
RUN_ID = "gc-2026-08-01-catchup"UNITS = set("ABCDEFG")CLUSTERS = {f"C{i}" for i in range(1, 14)}REQUIRED = ("manifest.txt", "scope.md", "workflow.dot", "events.log", "gradebook.jsonl", "clusters.jsonl", "deferrals.jsonl", "todos.jsonl")EVENTS = {"setup", "raw_discovered", "clustered", "verified", "rejected", "needs_more_evidence", "cleanup_ready", "planned", "in_progress", "cleaned", "partially_cleaned", "deferred", "blocked", "ejected_from_gc", "accepted_risk", "stale_expired"}TERMINAL = {"done", "blocked", "deferred", "rejected", "ejected_from_gc", "partially_cleaned"}OPERATOR_GATE_TODOS = {44, 45, 46, 47}POST_ACCEPTANCE_TODOS = {48, 49, 50}
def fail(message: str) -> None: raise ValueError(message)
def read_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): fail(f"missing {path}") records: list[dict[str, Any]] = [] for line_no, line in enumerate(path.read_text().splitlines(), 1): if not line.strip(): continue try: value = json.loads(line) except json.JSONDecodeError as exc: fail(f"{path}:{line_no}: invalid JSON: {exc}") if not isinstance(value, dict): fail(f"{path}:{line_no}: record is not an object") records.append(value) return records
def validate_event(record: dict[str, Any], source: str) -> None: for field in ("schema_version", "run_id", "event", "at", "by", "notes"): if field not in record: fail(f"{source}: event missing {field}") if record["schema_version"] != 1 or record["run_id"] != RUN_ID: fail(f"{source}: wrong schema version or run id") if record["event"] not in EVENTS: fail(f"{source}: unknown event {record['event']!r}") if not isinstance(record["at"], str) or not record["at"]: fail(f"{source}: event has no timestamp") if not isinstance(record["by"], str) or not record["by"]: fail(f"{source}: event has no actor/session") if not isinstance(record["notes"], str) or not record["notes"]: fail(f"{source}: event has no notes") if "cluster_id" in record and record["cluster_id"] not in CLUSTERS: fail(f"{source}: unknown cluster {record['cluster_id']!r}") if "cleanup_unit" in record and record["cleanup_unit"] not in UNITS: fail(f"{source}: unknown cleanup unit {record['cleanup_unit']!r}") if record["event"] in {"blocked", "deferred", "rejected", "ejected_from_gc"} and not record.get("evidence"): fail(f"{source}: {record['event']} needs evidence") if record["event"] in {"cleaned", "partially_cleaned"} and not record.get("evidence"): fail(f"{source}: terminal cleanup event needs evidence")
def validate_state(root: Path, require_dispositions: bool, require_reviews: bool) -> None: state = root / ".current-gc" if not state.is_dir(): fail(".current-gc is missing") for rel in REQUIRED: if not (state / rel).is_file(): fail(f"missing .current-gc/{rel}") for unit in UNITS: for rel in (f"cleanup-units/{unit}.md", f"plans/plan-{unit}.md", f"findings/findings-{unit}.jsonl"): if not (state / rel).is_file(): fail(f"missing .current-gc/{rel}") if not (state / "findings/findings-final.jsonl").is_file(): fail("missing .current-gc/findings/findings-final.jsonl") gitignore = (root / ".gitignore").read_text() if (root / ".gitignore").exists() else "" if not re.search(r"(?m)^/?\.current-gc/?$", gitignore): fail(".gitignore does not contain the exact .current-gc/ rule") if f"run_id={RUN_ID}" not in (state / "manifest.txt").read_text(): fail("manifest does not name the run id") events: list[dict[str, Any]] = [] for rel in ("events.log", "gradebook.jsonl"): records = read_jsonl(state / rel) for index, record in enumerate(records, 1): validate_event(record, f".current-gc/{rel}:{index}") events.extend(records) clusters = read_jsonl(state / "clusters.jsonl") seen_clusters: set[str] = set() for index, record in enumerate(clusters, 1): if record.get("run_id") != RUN_ID or record.get("schema_version") != 1: fail(f"clusters.jsonl:{index}: wrong run metadata") cluster = record.get("cluster_id") if cluster not in CLUSTERS: fail(f"clusters.jsonl:{index}: invalid cluster") seen_clusters.add(cluster) if seen_clusters != CLUSTERS: fail(f"clusters.jsonl does not cover all clusters: {sorted(CLUSTERS - seen_clusters)}") deferrals = read_jsonl(state / "deferrals.jsonl") for index, record in enumerate(deferrals, 1): for field in ("cluster_id", "disposition", "rationale", "evidence", "follow_up_condition", "owner"): if not record.get(field): fail(f"deferrals.jsonl:{index}: missing {field}") if record["cluster_id"] not in CLUSTERS or record["owner"] != "future implementation": fail(f"deferrals.jsonl:{index}: invalid deferral") if not {"C3", "C4", "C7", "C12"}.issubset({record["cluster_id"] for record in deferrals}): fail("required C3/C4/C7/C12 deferrals are missing") todos = read_jsonl(state / "todos.jsonl") if not todos: fail("todos.jsonl is empty") for index, record in enumerate(todos, 1): for field in ("id", "title", "status", "dependencies", "checkpoint_evidence"): if field not in record: fail(f"todos.jsonl:{index}: missing {field}") if record["status"] not in TERMINAL | {"pending", "in_progress"}: fail(f"todos.jsonl:{index}: invalid status") todo_id = int(record["id"]) if record.get("operator_gate") and todo_id not in OPERATOR_GATE_TODOS: fail(f"todos.jsonl:{index}: operator_gate is not allowed for todo {todo_id}") if record.get("post_acceptance") and todo_id not in POST_ACCEPTANCE_TODOS: fail(f"todos.jsonl:{index}: post_acceptance is not allowed for todo {todo_id}") if require_dispositions: event_clusters = {record.get("cluster_id") for record in events} for cluster in ("C3", "C4", "C5", "C7", "C12", "C13"): if cluster not in event_clusters: fail(f"required disposition event missing for {cluster}") dispositions = {record.get("cluster_id"): record["event"] for record in events if record["event"] in {"deferred", "blocked", "rejected", "ejected_from_gc"}} for cluster in ("C3", "C4", "C7", "C12"): if dispositions.get(cluster) not in {"deferred", "blocked"}: fail(f"{cluster} has no deferred/blocked disposition") if dispositions.get("C5") not in {"rejected", "ejected_from_gc"} or dispositions.get("C13") not in {"rejected", "ejected_from_gc"}: fail("C5/C13 must be rejected or ejected") if require_reviews: findings = [] for rel in [f"findings/findings-{unit}.jsonl" for unit in UNITS] + ["findings/findings-final.jsonl"]: findings.extend(read_jsonl(state / rel)) if not findings: fail("review records are required") for index, finding in enumerate(findings, 1): for field in ("finding_id", "severity", "disposition", "resolution"): if not finding.get(field): fail(f"review record {index} missing {field}") if finding["severity"] in {"Critical", "High"} and finding["disposition"] not in {"fixed", "rebutted", "accepted_risk"}: fail(f"unresolved {finding['severity']} finding {finding['finding_id']}") if any(record["status"] not in TERMINAL and not record.get("operator_gate") and not record.get("post_acceptance") for record in todos): fail("all pre-acceptance todos must be terminal for --require-review-records") print(f"GC state valid: {RUN_ID}")
def self_test() -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) state = root / ".current-gc" state.mkdir() (root / ".gitignore").write_text(".current-gc/\n") for rel in REQUIRED: (state / rel).write_text(f"run_id={RUN_ID}\n" if rel == "manifest.txt" else "") for unit in UNITS: for rel in (f"cleanup-units/{unit}.md", f"plans/plan-{unit}.md", f"findings/findings-{unit}.jsonl"): path = state / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_text("" if path.suffix == ".jsonl" else "fixture") final = state / "findings/findings-final.jsonl" final.parent.mkdir(parents=True, exist_ok=True) final.write_text("") event = {"schema_version": 1, "run_id": RUN_ID, "event": "verified", "at": "2026-08-01T00:00:00Z", "by": "self-test", "notes": "fixture"} (state / "events.log").write_text(json.dumps(event) + "\n") (state / "gradebook.jsonl").write_text(json.dumps(event) + "\n") (state / "clusters.jsonl").write_text("\n".join(json.dumps({"schema_version": 1, "run_id": RUN_ID, "cluster_id": f"C{i}", "disposition": "verified"}) for i in range(1, 14)) + "\n") (state / "deferrals.jsonl").write_text("\n".join(json.dumps({"cluster_id": cluster, "disposition": "deferred", "rationale": "fixture", "evidence": ["fixture"], "follow_up_condition": "future", "owner": "future implementation"}) for cluster in ("C3", "C4", "C7", "C12")) + "\n") (state / "todos.jsonl").write_text(json.dumps({"id": "1", "title": "fixture", "status": "pending", "dependencies": [], "checkpoint_evidence": ["fixture"]}) + "\n") validate_state(root, False, False) (state / "events.log").write_text("not json\n") try: validate_state(root, False, False) except ValueError: pass else: fail("negative fixture unexpectedly passed")
(state / "events.log").write_text(json.dumps(event) + "\n") valid_todo = { "id": "44", "title": "operator acceptance", "status": "pending", "dependencies": [], "checkpoint_evidence": [], "operator_gate": True, } (state / "todos.jsonl").write_text(json.dumps(valid_todo) + "\n") validate_state(root, False, False) invalid_todo = {**valid_todo, "id": "1"} (state / "todos.jsonl").write_text(json.dumps(invalid_todo) + "\n") try: validate_state(root, False, False) except ValueError: pass else: fail("ordinary todo gate fixture unexpectedly passed") print("validate-gc-state self-test passed")
def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--self-test", action="store_true") parser.add_argument("--require-dispositions", action="store_true") parser.add_argument("--require-review-records", action="store_true") parser.add_argument("run_id", nargs="?", default=RUN_ID) args = parser.parse_args() try: if args.run_id != RUN_ID: fail(f"unsupported run id {args.run_id}") if args.self_test: self_test() else: validate_state(Path.cwd(), args.require_dispositions, args.require_review_records) except ValueError as exc: print(f"GC state invalid: {exc}", file=sys.stderr) return 1 return 0
if __name__ == "__main__": raise SystemExit(main())