546 lines
22 KiB
Python
546 lines
22 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, 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 <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()
|