196 lines
7.5 KiB
Python
196 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate work-record YAML blocks against the canon kind registry.
|
|
|
|
Canon: canon/standards/work-record-types_v0.1.md (+ work-record-types.yaml)
|
|
Schemas: canon/standards/schemas/work-records/*.schema.json
|
|
|
|
Scans Markdown files for fenced ```yaml and ```task blocks. A block is a
|
|
work record iff it parses to a mapping whose `id` matches a registered (or
|
|
grandfathered legacy) id pattern. Records are validated against their
|
|
kind's JSON schema; ``task`` blocks get pattern + lifecycle checks (they
|
|
are already parsed authoritatively by state-hub).
|
|
|
|
Blocks with an id-like string that matches *no* registered pattern produce
|
|
a warning (CI-level sidetrack hint; the authoritative detector is
|
|
fix-consistency C-31).
|
|
|
|
Usage:
|
|
validate_work_records.py [--repo PATH] [--canon PATH] [--strict]
|
|
|
|
--strict turns sidetrack warnings into errors. Exit 1 on any error.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
FENCE_RE = re.compile(r"```(yaml|task)\n(.*?)```", re.S)
|
|
ID_LIKE_RE = re.compile(r"^[A-Z][A-Z0-9]*(-[A-Z0-9]+)+-?[0-9]*$")
|
|
SKIP_DIRS = {".git", "node_modules", ".venv", "history", "agents_backup"}
|
|
TERMINAL = {"closed", "resolved", "done", "cancel", "finished", "archived"}
|
|
|
|
# task blocks are validated inline (state-hub owns their full parsing). The
|
|
# registry is the sole id-pattern authority, including grandfathered schemes.
|
|
TASK_STATUS = {"wait", "todo", "progress", "done", "cancel",
|
|
"in_progress", "blocked"} # aliases per migration window
|
|
|
|
|
|
def load_registry(canon: Path) -> list[dict]:
|
|
reg = yaml.safe_load(
|
|
(canon / "canon/standards/work-record-types.yaml").read_text())
|
|
kinds = []
|
|
for k in reg["kinds"]:
|
|
pats = [re.compile(p) for p in k.get("id_patterns", [])]
|
|
pats += [re.compile(lp["pattern"])
|
|
for lp in k.get("legacy_patterns", [])]
|
|
kinds.append({"kind": k["kind"], "patterns": pats})
|
|
return kinds
|
|
|
|
|
|
def load_validators(canon: Path):
|
|
"""Return {kind: callable(block) -> list[str]} using jsonschema if
|
|
available, else minimal fallback checks."""
|
|
schema_dir = canon / "canon/standards/schemas/work-records"
|
|
schemas = {p.stem.replace(".schema", ""): json.loads(p.read_text())
|
|
for p in schema_dir.glob("*.schema.json")}
|
|
try:
|
|
import jsonschema
|
|
|
|
spine_defs = schemas["spine"]["$defs"]
|
|
|
|
def inline(node):
|
|
"""Replace spine.schema.json#/$defs/X refs with the def body
|
|
so no ref registry is needed (works on any jsonschema>=4)."""
|
|
if isinstance(node, dict):
|
|
ref = node.get("$ref", "")
|
|
if ref.startswith("spine.schema.json#/$defs/"):
|
|
merged = dict(spine_defs[ref.rsplit("/", 1)[1]])
|
|
merged.update({k: v for k, v in node.items()
|
|
if k != "$ref"})
|
|
return inline(merged)
|
|
return {k: inline(v) for k, v in node.items()}
|
|
if isinstance(node, list):
|
|
return [inline(v) for v in node]
|
|
return node
|
|
|
|
def make(kind):
|
|
validator = jsonschema.Draft202012Validator(
|
|
inline(schemas[kind]))
|
|
|
|
def check(block):
|
|
return [f"{e.json_path}: {e.message}"
|
|
for e in validator.iter_errors(block)]
|
|
return check
|
|
|
|
return {k: make(k) for k in schemas if k != "spine"}
|
|
except ImportError:
|
|
def fallback(kind):
|
|
def check(block):
|
|
errs = []
|
|
if not block.get("title") and kind != "task":
|
|
errs.append("missing title (fallback check)")
|
|
if block.get("status") not in TERMINAL and \
|
|
"lane" not in block and kind == "intake":
|
|
errs.append("open record missing lane (fallback check)")
|
|
return errs
|
|
return check
|
|
print("note: jsonschema not installed — minimal fallback checks only",
|
|
file=sys.stderr)
|
|
return {k: fallback(k) for k in schemas if k != "spine"}
|
|
|
|
|
|
def classify(record_id: str, kinds: list[dict]) -> str | None:
|
|
for k in kinds:
|
|
if any(p.match(record_id) for p in k["patterns"]):
|
|
return k["kind"]
|
|
return None
|
|
|
|
|
|
def iter_blocks(md: Path, kinds: list[dict]):
|
|
for fence, body in FENCE_RE.findall(md.read_text(errors="replace")):
|
|
try:
|
|
docs = list(yaml.safe_load_all(body))
|
|
except yaml.YAMLError as exc:
|
|
# only an error if the raw text plausibly holds a registered id
|
|
raw_ids = re.findall(r"^id:\s*(\S+)", body, re.M)
|
|
if any(classify(r.strip("\"'"), kinds) for r in raw_ids):
|
|
yield fence, None, f"unparseable YAML block: {exc}"
|
|
continue
|
|
for data in docs:
|
|
if isinstance(data, dict):
|
|
yield fence, data, None
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--repo", default=".", type=Path)
|
|
ap.add_argument("--canon", default=None, type=Path,
|
|
help="the-custodian checkout (default: script's repo)")
|
|
ap.add_argument("--strict", action="store_true")
|
|
args = ap.parse_args()
|
|
canon = args.canon or Path(__file__).resolve().parents[1]
|
|
|
|
kinds = load_registry(canon)
|
|
validators = load_validators(canon)
|
|
errors, warnings, checked = [], [], 0
|
|
|
|
md_files = [p for p in sorted(args.repo.rglob("*.md"))
|
|
if not any(part in SKIP_DIRS for part in p.parts)]
|
|
for md in md_files:
|
|
rel = md.relative_to(args.repo)
|
|
for fence, block, parse_err in iter_blocks(md, kinds):
|
|
if parse_err:
|
|
errors.append(f"{rel}: {parse_err}")
|
|
continue
|
|
rid = block.get("id")
|
|
if not isinstance(rid, str):
|
|
continue
|
|
if "NNN" in rid or rid.endswith("-TNN"):
|
|
continue # template placeholder, not a record
|
|
kind = classify(rid, kinds)
|
|
if kind is None:
|
|
if ID_LIKE_RE.match(rid):
|
|
warnings.append(
|
|
f"{rel}: '{rid}' looks like a work-record id but "
|
|
f"matches no registered pattern — unregistered "
|
|
f"species are sidetracks (work-record-types_v0.1)")
|
|
continue
|
|
checked += 1
|
|
if fence == "task" and kind != "task":
|
|
errors.append(
|
|
f"{rel}: task fence id '{rid}' is registered as {kind}, "
|
|
"not task"
|
|
)
|
|
continue
|
|
if kind == "task":
|
|
st = block.get("status")
|
|
if st is not None and st not in TASK_STATUS:
|
|
errors.append(f"{rel}: {rid}: bad task status '{st}'")
|
|
continue
|
|
if kind in validators:
|
|
for msg in validators[kind](block):
|
|
# historical grace: terminal records only get id/status
|
|
# structural errors, not spine completeness
|
|
if block.get("status") in TERMINAL and "required" in msg:
|
|
continue
|
|
errors.append(f"{rel}: {rid}: {msg}")
|
|
|
|
for w in warnings:
|
|
print(f"WARN {w}")
|
|
for e in errors:
|
|
print(f"ERROR {e}")
|
|
print(f"work-records: {checked} checked, "
|
|
f"{len(errors)} errors, {len(warnings)} warnings")
|
|
if errors or (args.strict and warnings):
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|