367 lines
14 KiB
Python
367 lines
14 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
validate_report.py — Machine-report schema validator for doc-hygiene.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
python validate_report.py <path-to-report.json>
|
||
|
|
|
||
|
|
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
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Safety-tier derivation (invariant #10)
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
def derive_safety_tier(op_type: str, is_destructive: bool, is_reversible: bool) -> str:
|
||
|
|
"""
|
||
|
|
Deterministic derivation of safety_tier from op characterisation.
|
||
|
|
Never returns 'auto' for a generative, destructive, or irreversible op.
|
||
|
|
"""
|
||
|
|
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,
|
||
|
|
),
|
||
|
|
}
|
||
|
|
|
||
|
|
STALE_SUBTYPES = {"contradicted", "orphaned", "superseded", "provisional", "completed-in-place", "duplicated"}
|
||
|
|
BLOAT_SUBTYPES = {"distill", "split", "freeze"}
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# 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")
|
||
|
|
|
||
|
|
return self._violations
|
||
|
|
|
||
|
|
def _check_envelope(self, r: dict) -> None:
|
||
|
|
for key in ("schema_version", "tool_version", "generated_at", "scan", "shortlist", "entries"):
|
||
|
|
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")
|
||
|
|
|
||
|
|
# Rules 6, 7, 8, 9: exact_edit checks (only if op_type valid)
|
||
|
|
if op_type in ("deterministic", "generative"):
|
||
|
|
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)
|
||
|
|
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})",
|
||
|
|
)
|
||
|
|
|
||
|
|
# 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_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) -> None:
|
||
|
|
"""Rule 6: exact_edit present IFF op_type == deterministic."""
|
||
|
|
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 # 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
|
||
|
|
|
||
|
|
if op_type == "deterministic" and has_exact_edit:
|
||
|
|
self._check_exact_edit(exact_edit, pfx, entry)
|
||
|
|
|
||
|
|
def _check_exact_edit(self, exact_edit: Any, pfx: str, entry: dict) -> None:
|
||
|
|
"""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
|
||
|
|
|
||
|
|
kind = exact_edit.get("kind")
|
||
|
|
if kind is None:
|
||
|
|
self._add(f"{pfx}.exact_edit.kind", "exact_edit.kind is required")
|
||
|
|
return
|
||
|
|
|
||
|
|
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
|
||
|
|
|
||
|
|
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}'",
|
||
|
|
)
|
||
|
|
|
||
|
|
# 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}",
|
||
|
|
)
|
||
|
|
|
||
|
|
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 <path-to-report.json>"}))
|
||
|
|
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()
|