#!/usr/bin/env python3 """ validate_report.py — Machine-report schema validator for doc-hygiene. Usage: python validate_report.py Exit codes: 0 — report is valid 1 — report is invalid (violations reported in JSON output) 2 — usage error (missing argument, file not found, not JSON) Output: structured JSON on stdout with keys: valid: bool violations: list of {field, message} """ import json import sys from dataclasses import dataclass, field from typing import Any, Optional # --------------------------------------------------------------------------- # Safety-tier derivation (invariant #10) # --------------------------------------------------------------------------- def _lifecycle_is_scanner_proven(lifecycle: dict) -> bool: """True when the lifecycle evidence is a filesystem-provable fact. A `temporary`-lifetime match is scanner-computed (retain-recent/age); a `served_when_path` hit is scanner-verified against the filesystem. A `served_when` (free text) is always classifier-judged, never proven. """ if lifecycle.get("lifetime") == "temporary": return True if lifecycle.get("served_when_path"): return True return False def derive_safety_tier( op_type: str, is_destructive: bool, is_reversible: bool, lifecycle: "dict | None" = None, ) -> str: """ Deterministic derivation of safety_tier from op characterisation. Never returns 'auto' for a generative, destructive, or irreversible op. ADR-0039 lifecycle branch: when `lifecycle` is supplied (i.e. this is a `delete`/`extract-then-delete` op), the tier is derived from evidence quality + runtime git state instead of (is_destructive, is_reversible) — `auto` only when the lifecycle evidence is scanner-proven AND the path is tracked AND clean; every other combination (dirty, untracked, or classifier-judged `served_when`) forces `confirm`, regardless of the is_destructive/is_reversible inputs. This is additive: the existing non-lifecycle branches below are unchanged. """ if lifecycle is not None: git_state = lifecycle.get("git_state") or {} tracked = bool(git_state.get("tracked")) clean = bool(git_state.get("clean")) if _lifecycle_is_scanner_proven(lifecycle) and tracked and clean: return "auto" return "confirm" if op_type == "generative": return "confirm" if is_destructive: return "confirm" if not is_reversible: return "confirm" return "auto" # --------------------------------------------------------------------------- # Kind table: closed enum + per-kind required fields + inherent booleans # --------------------------------------------------------------------------- @dataclass class KindSpec: is_destructive: bool is_reversible: bool required_fields: list # fields required directly on exact_edit (besides kind) has_anchor: bool # whether anchor sub-object is required KIND_TABLE: dict[str, KindSpec] = { "delete-range": KindSpec( is_destructive=True, is_reversible=False, required_fields=["anchor"], has_anchor=True, ), "move-to-archive": KindSpec( is_destructive=False, is_reversible=True, required_fields=["anchor", "dest_path"], has_anchor=True, ), "insert-frontmatter": KindSpec( is_destructive=False, is_reversible=True, required_fields=["key", "value"], has_anchor=False, ), "replace-text": KindSpec( is_destructive=False, is_reversible=True, required_fields=["anchor", "match", "replacement"], has_anchor=True, ), "dedupe": KindSpec( is_destructive=False, is_reversible=True, required_fields=["anchor", "canonical_ref"], has_anchor=True, ), "delete": KindSpec( # is_destructive is fixed (git history is the only recovery path); # is_reversible is a nominal True here — its REAL characterisation is # git-history-dependent and is resolved by the lifecycle branch of # derive_safety_tier, not by this fixed table value (report-schema # spec, "Exact-Edit Kind Is a Closed Enum"). is_destructive=True, is_reversible=True, required_fields=["anchor", "lifecycle"], has_anchor=True, ), "extract-then-delete": KindSpec( is_destructive=True, is_reversible=True, required_fields=["anchor", "lifecycle", "extraction_target"], has_anchor=True, ), } STALE_SUBTYPES = {"contradicted", "orphaned", "superseded", "provisional", "completed-in-place", "duplicated"} BLOAT_SUBTYPES = {"distill", "split", "freeze"} _LIFECYCLE_LIFETIMES = {"keep", "temporary", "delete-once-served"} _LIFECYCLE_OP_KINDS = {"delete", "extract-then-delete"} _EXTRACTION_TARGETS = {"repo-durable", "cross-repo"} # --------------------------------------------------------------------------- # Validator # --------------------------------------------------------------------------- class ReportValidator: """Validates a machine-report JSON dict against the frozen schema contract.""" def __init__(self, report: Any): self._report = report self._violations: list[dict] = [] def _add(self, field_path: str, message: str) -> None: self._violations.append({"field": field_path, "message": message}) def validate(self) -> list[dict]: """Run all checks; collect all violations. Returns list of violations.""" r = self._report if not isinstance(r, dict): self._add("(root)", "Report must be a JSON object") return self._violations # Rule 1: Envelope fields self._check_envelope(r) # Rule 2: shortlist precedes entries shortlist = r.get("shortlist", []) if not isinstance(shortlist, list): self._add("shortlist", "Must be an array") shortlist = [] entries = r.get("entries", []) if not isinstance(entries, list): self._add("entries", "Must be an array") entries = [] shortlist_set = set(shortlist) if isinstance(shortlist, list) else set() for i, entry in enumerate(entries): if isinstance(entry, dict): self._check_entry(entry, i, shortlist_set) else: self._add(f"entries[{i}]", "Entry must be a JSON object") # Rule: promotion_candidates section (report-schema delta spec). if "promotion_candidates" in r: self._check_promotion_candidates(r["promotion_candidates"]) return self._violations def _check_envelope(self, r: dict) -> None: for key in ( "schema_version", "tool_version", "generated_at", "scan", "shortlist", "entries", "promotion_candidates", ): if key not in r: self._add(key, f"Required top-level field '{key}' is missing") scan = r.get("scan") if scan is not None: if not isinstance(scan, dict): self._add("scan", "Must be a JSON object") else: for key in ("project_root", "scope_globs", "excluded_dirs", "files_scanned"): if key not in scan: self._add(f"scan.{key}", f"Required scan field '{key}' is missing") def _check_entry(self, entry: dict, idx: int, shortlist_set: set) -> None: pfx = f"entries[{idx}]" # Rule 3: Required entry fields required = ( "path", "category", "signals", "op", "op_type", "is_destructive", "is_reversible", "safety_tier", "token_estimate", ) for f in required: if f not in entry: self._add(f"{pfx}.{f}", f"Required entry field '{f}' is missing") # Rule 2: entry path must be in shortlist path = entry.get("path") if path is not None and shortlist_set and path not in shortlist_set: self._add(f"{pfx}.path", f"Entry path '{path}' is not in shortlist") # Rule 4: Category closed enum category = entry.get("category") if category is not None: self._check_category(category, pfx) # Rule 5: Scalar enums op_type = entry.get("op_type") if op_type is not None and op_type not in ("deterministic", "generative"): self._add(f"{pfx}.op_type", f"op_type must be 'deterministic' or 'generative', got '{op_type}'") safety_tier = entry.get("safety_tier") if safety_tier is not None and safety_tier not in ("auto", "confirm"): self._add(f"{pfx}.safety_tier", f"safety_tier must be 'auto' or 'confirm', got '{safety_tier}'") is_destructive = entry.get("is_destructive") if is_destructive is not None and not isinstance(is_destructive, bool): self._add(f"{pfx}.is_destructive", "is_destructive must be a boolean") is_reversible = entry.get("is_reversible") if is_reversible is not None and not isinstance(is_reversible, bool): self._add(f"{pfx}.is_reversible", "is_reversible must be a boolean") # entry-level lifecycle signal (report-schema: "Lifecycle Signal # Fields on Shortlist and Entries") — optional; validated when present, # regardless of op kind (a `keep`/`temporary` entry may carry it with # no exact_edit at all). if "lifecycle" in entry: self._check_lifecycle(entry["lifecycle"], f"{pfx}.lifecycle", require_git_state=False) # Rules 6, 7, 8, 9: exact_edit checks (only if op_type valid) entry_lifecycle_for_tier = None if op_type in ("deterministic", "generative"): entry_lifecycle_for_tier = self._check_exact_edit_biconditional(entry, pfx, op_type) # Rule 9: safety_tier derivation match (only when all inputs are valid) if ( op_type in ("deterministic", "generative") and isinstance(is_destructive, bool) and isinstance(is_reversible, bool) and safety_tier in ("auto", "confirm") ): expected_tier = derive_safety_tier( op_type, is_destructive, is_reversible, lifecycle=entry_lifecycle_for_tier ) if safety_tier != expected_tier: self._add( f"{pfx}.safety_tier", f"safety_tier mismatch: recorded '{safety_tier}' but derivation yields '{expected_tier}' " f"for (op_type={op_type}, is_destructive={is_destructive}, is_reversible={is_reversible}, " f"lifecycle={'present' if entry_lifecycle_for_tier is not None else 'none'})", ) # Rule 10: token_estimate.raw_tokens required token_estimate = entry.get("token_estimate") if token_estimate is not None: self._check_token_estimate(token_estimate, pfx) def _check_promotion_candidates(self, candidates: Any) -> None: if not isinstance(candidates, list): self._add("promotion_candidates", "promotion_candidates must be an array") return for i, cand in enumerate(candidates): pfx = f"promotion_candidates[{i}]" if not isinstance(cand, dict): self._add(pfx, "candidate must be a JSON object") continue for field_name in ("path", "name", "pitch"): value = cand.get(field_name) if not isinstance(value, str) or not value: self._add(f"{pfx}.{field_name}", f"candidate '{field_name}' must be a non-empty string") def _check_lifecycle( self, lifecycle: Any, pfx: str, *, require_git_state: bool ) -> Optional[dict]: """Validate a lifecycle object's shape (rule_ref, lifetime, exactly one served field per lifetime, optional/required git_state). Returns the lifecycle dict when it is well-formed enough to feed `derive_safety_tier` (or None if too malformed to trust). """ if not isinstance(lifecycle, dict): self._add(pfx, "lifecycle must be a JSON object") return None rule_ref = lifecycle.get("rule_ref") if not isinstance(rule_ref, str) or not rule_ref: self._add(f"{pfx}.rule_ref", "lifecycle.rule_ref is required and must be a non-empty string") lifetime = lifecycle.get("lifetime") if lifetime not in _LIFECYCLE_LIFETIMES: self._add( f"{pfx}.lifetime", f"lifecycle.lifetime must be one of {sorted(_LIFECYCLE_LIFETIMES)}, got {lifetime!r}", ) has_served_when_path = "served_when_path" in lifecycle and lifecycle["served_when_path"] is not None has_served_when = "served_when" in lifecycle and lifecycle["served_when"] is not None if lifetime == "delete-once-served": if has_served_when_path and has_served_when: self._add( f"{pfx}", "lifecycle must carry exactly one of served_when_path or served_when, not both", ) elif not has_served_when_path and not has_served_when: self._add( f"{pfx}", "lifecycle with lifetime 'delete-once-served' requires exactly one of " "served_when_path or served_when", ) elif lifetime in ("keep", "temporary"): if has_served_when_path or has_served_when: self._add( f"{pfx}", f"lifecycle with lifetime {lifetime!r} must not carry served_when_path or served_when " "(age/keep entries are not served-keyed)", ) if require_git_state: git_state = lifecycle.get("git_state") if not isinstance(git_state, dict): self._add(f"{pfx}.git_state", "lifecycle.git_state is required for delete/extract-then-delete") else: for k in ("tracked", "clean"): if not isinstance(git_state.get(k), bool): self._add(f"{pfx}.git_state.{k}", f"lifecycle.git_state.{k} must be a boolean") return lifecycle def _check_category(self, category: Any, pfx: str) -> None: if not isinstance(category, dict): self._add(f"{pfx}.category", "category must be a JSON object with 'class' and 'subtype'") return cls = category.get("class") subtype = category.get("subtype") if cls is None: self._add(f"{pfx}.category.class", "category.class is required") elif cls not in ("stale", "bloat"): self._add(f"{pfx}.category.class", f"category.class must be 'stale' or 'bloat', got '{cls}'") if subtype is None: self._add(f"{pfx}.category.subtype", "category.subtype is required") elif cls == "stale" and subtype not in STALE_SUBTYPES: self._add( f"{pfx}.category.subtype", f"For class 'stale', subtype must be one of {sorted(STALE_SUBTYPES)}, got '{subtype}'", ) elif cls == "bloat" and subtype not in BLOAT_SUBTYPES: self._add( f"{pfx}.category.subtype", f"For class 'bloat', subtype must be one of {sorted(BLOAT_SUBTYPES)}, got '{subtype}'", ) def _check_exact_edit_biconditional(self, entry: dict, pfx: str, op_type: str) -> Optional[dict]: """Rule 6: exact_edit present IFF op_type == deterministic. Returns the exact_edit's validated `lifecycle` dict (for delete / extract-then-delete kinds) so the caller can feed it to `derive_safety_tier`; None otherwise. """ has_exact_edit = "exact_edit" in entry exact_edit = entry.get("exact_edit") if op_type == "generative" and has_exact_edit: self._add( f"{pfx}.exact_edit", "generative entry must NOT carry exact_edit (invariant #11)", ) return None # No point validating the structure of an edit that shouldn't exist if op_type == "deterministic" and not has_exact_edit: self._add( f"{pfx}.exact_edit", "deterministic entry MUST carry exact_edit (invariant #11)", ) return None if op_type == "deterministic" and has_exact_edit: return self._check_exact_edit(exact_edit, pfx, entry) return None def _check_exact_edit(self, exact_edit: Any, pfx: str, entry: dict) -> Optional[dict]: """Rules 7 and 8: kind enum, per-kind required fields, characterisation match.""" if not isinstance(exact_edit, dict): self._add(f"{pfx}.exact_edit", "exact_edit must be a JSON object") return None kind = exact_edit.get("kind") if kind is None: self._add(f"{pfx}.exact_edit.kind", "exact_edit.kind is required") return None if kind not in KIND_TABLE: self._add( f"{pfx}.exact_edit.kind", f"Unknown exact_edit.kind '{kind}'; valid kinds: {sorted(KIND_TABLE)}", ) return None spec = KIND_TABLE[kind] # Rule 7: Per-kind required sub-fields for req_field in spec.required_fields: if req_field not in exact_edit: self._add( f"{pfx}.exact_edit.{req_field}", f"exact_edit of kind '{kind}' requires field '{req_field}'", ) lifecycle_for_tier: Optional[dict] = None if kind in _LIFECYCLE_OP_KINDS: if "lifecycle" in exact_edit: lifecycle_for_tier = self._check_lifecycle( exact_edit["lifecycle"], f"{pfx}.exact_edit.lifecycle", require_git_state=True ) if kind == "extract-then-delete": self._check_extraction_target(exact_edit, pfx) # Anchor must have start_line and end_line if spec.has_anchor and "anchor" in exact_edit: anchor = exact_edit["anchor"] if not isinstance(anchor, dict): self._add(f"{pfx}.exact_edit.anchor", "anchor must be a JSON object") else: for ak in ("start_line", "end_line"): if ak not in anchor: self._add(f"{pfx}.exact_edit.anchor.{ak}", f"anchor.{ak} is required for kind '{kind}'") # expected_sha256 required for anchor-bearing kinds if spec.has_anchor and "expected_sha256" not in exact_edit: self._add( f"{pfx}.exact_edit.expected_sha256", f"exact_edit of kind '{kind}' (anchor-bearing) requires 'expected_sha256'", ) # Rule 8: Kind characterisation match — entry's booleans must equal kind's inherent pair entry_is_destructive = entry.get("is_destructive") entry_is_reversible = entry.get("is_reversible") if isinstance(entry_is_destructive, bool) and entry_is_destructive != spec.is_destructive: self._add( f"{pfx}.is_destructive", f"kind='{kind}' requires is_destructive={spec.is_destructive}, got {entry_is_destructive}", ) if isinstance(entry_is_reversible, bool) and entry_is_reversible != spec.is_reversible: self._add( f"{pfx}.is_reversible", f"kind='{kind}' requires is_reversible={spec.is_reversible}, got {entry_is_reversible}", ) return lifecycle_for_tier def _check_extraction_target(self, exact_edit: dict, pfx: str) -> None: target = exact_edit.get("extraction_target") if target not in _EXTRACTION_TARGETS: self._add( f"{pfx}.exact_edit.extraction_target", f"extraction_target must be one of {sorted(_EXTRACTION_TARGETS)}, got {target!r}", ) return if target == "repo-durable": dest = exact_edit.get("extraction_dest") if not isinstance(dest, str) or not dest: self._add( f"{pfx}.exact_edit.extraction_dest", "extraction_target 'repo-durable' requires a non-empty 'extraction_dest'", ) def _check_token_estimate(self, token_estimate: Any, pfx: str) -> None: if not isinstance(token_estimate, dict): self._add(f"{pfx}.token_estimate", "token_estimate must be a JSON object") return if "raw_tokens" not in token_estimate: self._add(f"{pfx}.token_estimate.raw_tokens", "token_estimate.raw_tokens is required in v1") else: raw = token_estimate["raw_tokens"] if not isinstance(raw, (int, float)): self._add(f"{pfx}.token_estimate.raw_tokens", "raw_tokens must be a number") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main() -> None: if len(sys.argv) != 2: print(json.dumps({"error": "Usage: validate_report.py "})) sys.exit(2) path = sys.argv[1] try: with open(path, "r", encoding="utf-8") as fh: report = json.load(fh) except FileNotFoundError: print(json.dumps({"error": f"File not found: {path}"})) sys.exit(2) except json.JSONDecodeError as exc: print(json.dumps({"error": f"Invalid JSON: {exc}"})) sys.exit(2) validator = ReportValidator(report) violations = validator.validate() result = { "valid": len(violations) == 0, "violations": violations, } print(json.dumps(result, indent=2)) sys.exit(0 if result["valid"] else 1) if __name__ == "__main__": main()