#!/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())