#!/usr/bin/env python3 """Validate the Avoid AI Writing cross-Skill orchestration graph. Stdlib only. Fails closed on dangling nodes/edges, missing connection contracts, unbounded cycles, terminal-node leaks, review-lens drift, legacy graph references, or a missing canonical fallback. """ from __future__ import annotations import json import sys from pathlib import Path ALLOWED_EDGE_TYPES = {"ROUTE", "FEED", "VERIFY", "REPAIR", "RECHECK", "ESCALATE"} SPECIALIZED = { "ai-writing-detector", "voice-preserving-rewriter", "file-edit-in-place", "preservation-verifier", "false-positive-reviewer", } EXPECTED_LENSES = { "agency-software-architect", "agency-ai-engineer", "agency-senior-developer", "agency-inclusive-visuals-specialist", } def fail(errors: list[str], message: str) -> None: errors.append(message) def typed_member( mapping: dict[str, object], key: str, expected_type: type, errors: list[str], label: str, default: object, ) -> object: """Return a typed mapping member, recording an error on type mismatch.""" value = mapping.get(key, default) if not isinstance(value, expected_type): kind = {dict: "object", list: "array", str: "string"}[expected_type] article = "an" if kind in {"object", "array"} else "a" fail(errors, f"{label} must be {article} {kind}") return default return value def string_list_member( mapping: dict[str, object], key: str, errors: list[str], label: str ) -> list[str]: """Return a string-array member, filtering invalid entries after reporting them.""" values = typed_member(mapping, key, list, errors, label, []) assert isinstance(values, list) strings: list[str] = [] for index, value in enumerate(values): if not isinstance(value, str): fail(errors, f"{label} item {index} must be a string") else: strings.append(value) return strings def first_cycle(adjacency: dict[str, list[str]]) -> list[str] | None: """Return one directed cycle if the graph contains one, otherwise None.""" state: dict[str, int] = {node: 0 for node in adjacency} stack: list[str] = [] positions: dict[str, int] = {} def visit(node: str) -> list[str] | None: state[node] = 1 positions[node] = len(stack) stack.append(node) for target in adjacency.get(node, []): if state.get(target, 0) == 0: cycle = visit(target) if cycle: return cycle elif state.get(target) == 1: start = positions[target] return stack[start:] + [target] stack.pop() positions.pop(node, None) state[node] = 2 return None for node in sorted(adjacency): if state[node] == 0: cycle = visit(node) if cycle: return cycle return None def main() -> int: root = Path(sys.argv[1] if len(sys.argv) > 1 else ".").resolve() refs = root / "skills/avoid-ai-writing-router/references" graph_path = refs / "skill-graph.json" handoff_path = refs / "handoff-contract.md" lenses_path = refs / "agency-role-lenses.md" routing_path = refs / "routing-matrix.md" router_path = root / "skills/avoid-ai-writing-router/SKILL.md" errors: list[str] = [] for path in (graph_path, handoff_path, lenses_path, routing_path, router_path): if not path.is_file(): fail(errors, f"missing required orchestration file: {path.relative_to(root)}") legacy = refs / "skill-graph.yaml" if legacy.exists(): fail(errors, "legacy skill-graph.yaml must not coexist with canonical skill-graph.json") if errors: for error in errors: print(f"ERROR: {error}") return 1 try: graph = json.loads(graph_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: print(f"ERROR: invalid skill graph: {exc}") return 1 if not isinstance(graph, dict): fail(errors, "skill graph JSON root must be an object") graph = {} if graph.get("version") != 2: fail(errors, "skill graph version must be 2") nodes = typed_member(graph, "nodes", dict, errors, "nodes", {}) assert isinstance(nodes, dict) if not nodes: fail(errors, "nodes must be a non-empty object") canonical = typed_member( graph, "canonical_authority", str, errors, "canonical_authority", "" ) entrypoint = typed_member(graph, "entrypoint", str, errors, "entrypoint", "") handoff_contract = typed_member( graph, "handoff_contract", str, errors, "handoff_contract", "" ) assert isinstance(canonical, str) assert isinstance(entrypoint, str) assert isinstance(handoff_contract, str) if canonical != "avoid-ai-writing": fail(errors, "canonical_authority must be avoid-ai-writing") if entrypoint != "avoid-ai-writing-router": fail(errors, "entrypoint must be avoid-ai-writing-router") if handoff_contract != "handoff-contract.md": fail(errors, "handoff_contract must point to handoff-contract.md") skills_root = root / "skills" if not skills_root.is_dir(): fail(errors, "skills directory is missing") skill_dirs: set[str] = set() else: skill_dirs = { p.name for p in skills_root.iterdir() if p.is_dir() and (p / "SKILL.md").is_file() } graph_nodes = set(nodes) missing_dirs = sorted(graph_nodes - skill_dirs) if missing_dirs: fail(errors, f"graph nodes without Skill directories: {missing_dirs}") uncovered_public_skills = sorted(skill_dirs - graph_nodes) if uncovered_public_skills: fail(errors, f"public Skills missing from orchestration graph: {uncovered_public_skills}") incoming: dict[str, int] = {name: 0 for name in graph_nodes} outgoing: dict[str, int] = {name: 0 for name in graph_nodes} unbounded_adjacency: dict[str, list[str]] = {name: [] for name in graph_nodes} edges = typed_member(graph, "edges", list, errors, "edges", []) assert isinstance(edges, list) seen_edges: set[tuple[str, str, str, str]] = set() for index, edge in enumerate(edges): if not isinstance(edge, dict): fail(errors, f"edge {index} must be an object") continue edge_type = typed_member(edge, "type", str, errors, f"edge {index} type", "") source = typed_member(edge, "from", str, errors, f"edge {index} from", "") target = typed_member(edge, "to", str, errors, f"edge {index} to", "") condition = typed_member(edge, "when", str, errors, f"edge {index} when", "") assert isinstance(edge_type, str) assert isinstance(source, str) assert isinstance(target, str) assert isinstance(condition, str) if edge_type not in ALLOWED_EDGE_TYPES: fail(errors, f"edge {index} has unsupported type: {edge_type!r}") if source not in graph_nodes: fail(errors, f"edge {index} has unknown source: {source!r}") if target not in graph_nodes: fail(errors, f"edge {index} has unknown target: {target!r}") if source == target and source: fail(errors, f"edge {index} creates a self-loop on {source}") if not condition.strip(): fail(errors, f"edge {index} requires a non-empty when condition") limit = edge.get("max_reentries") if limit is not None and (not isinstance(limit, int) or isinstance(limit, bool) or limit != 1): fail(errors, f"edge {source}->{target} has invalid max_reentries: {limit!r}") if edge_type in {"REPAIR", "RECHECK"} and limit != 1: fail(errors, f"{edge_type} edge {source}->{target} must set max_reentries to 1") if source in graph_nodes: outgoing[source] += 1 if target in graph_nodes: incoming[target] += 1 if source in graph_nodes and target in graph_nodes and limit != 1: unbounded_adjacency[source].append(target) key = (str(edge_type), str(source), str(target), str(condition)) if key in seen_edges: fail(errors, f"duplicate edge: {key}") seen_edges.add(key) loop_policy = typed_member(graph, "loop_policy", dict, errors, "loop_policy", {}) assert isinstance(loop_policy, dict) if not loop_policy: fail(errors, "loop_policy must be present") else: if loop_policy.get("canonical_rewrite_pass_max") != 2: fail(errors, "canonical_rewrite_pass_max must remain 2") if loop_policy.get("repair_reentry_max") != 1: fail(errors, "repair_reentry_max must remain 1") if loop_policy.get("self_loops_allowed") is not False: fail(errors, "self_loops_allowed must be false") if loop_policy.get("terminal_nodes_have_no_outgoing_edges") is not True: fail(errors, "terminal_nodes_have_no_outgoing_edges must be true") if loop_policy.get("every_graph_cycle_requires_bounded_edge") is not True: fail(errors, "every_graph_cycle_requires_bounded_edge must be true") for name in graph_nodes: node = nodes.get(name, {}) terminal = bool(node.get("terminal")) if isinstance(node, dict) else False if name not in {entrypoint, canonical} and incoming.get(name, 0) == 0: fail(errors, f"Skill has no incoming orchestration edge: {name}") if terminal and outgoing.get(name, 0) != 0: fail(errors, f"terminal Skill has outgoing orchestration edges: {name}") if not terminal and outgoing.get(name, 0) == 0: fail(errors, f"non-terminal Skill has no outgoing orchestration edge: {name}") reviewer_node = nodes.get("false-positive-reviewer", {}) if not isinstance(reviewer_node, dict) or reviewer_node.get("terminal") is not True: fail(errors, "false-positive-reviewer must remain terminal") else: router_reasons = set( string_list_member( reviewer_node, "return_control_to_router_when", errors, "false-positive-reviewer return_control_to_router_when", ) ) expected_reasons = {"fresh_signal_collection_needed", "intent_changes_to_rewrite_or_edit"} if router_reasons != expected_reasons: fail(errors, "false-positive-reviewer router-return reasons drifted") unbounded_cycle = first_cycle(unbounded_adjacency) if unbounded_cycle: fail(errors, "unbounded orchestration cycle detected: " + " -> ".join(unbounded_cycle)) fallback = typed_member(graph, "fallback", dict, errors, "fallback", {}) assert isinstance(fallback, dict) if fallback.get("skill") != canonical: fail(errors, "fallback must return to the canonical avoid-ai-writing Skill") lenses = set(string_list_member(graph, "review_lenses", errors, "review_lenses")) if lenses != EXPECTED_LENSES: fail(errors, f"review_lenses mismatch: expected {sorted(EXPECTED_LENSES)}, got {sorted(lenses)}") lenses_text = lenses_path.read_text(encoding="utf-8") for lens in EXPECTED_LENSES: if f"`{lens}`" not in lenses_text: fail(errors, f"agency-role-lenses.md is missing encoded lens: {lens}") guards = typed_member(graph, "guards", list, errors, "guards", []) assert isinstance(guards, list) if not guards: fail(errors, "at least one conditional guard is required") else: guard_names: set[str] = set() for index, guard in enumerate(guards): if not isinstance(guard, dict): fail(errors, f"guard {index} must be an object") continue name = typed_member(guard, "name", str, errors, f"guard {index} name", "") assert isinstance(name, str) guard_names.add(name) if "human_representation_preservation" not in guard_names: fail(errors, "human_representation_preservation guard is required") if "authorship_uncertainty" not in guard_names: fail(errors, "authorship_uncertainty guard is required") for index, guard in enumerate(guards): if not isinstance(guard, dict): continue lens = guard.get("review_lens") if lens is not None and not isinstance(lens, str): fail(errors, f"guard {index} review_lens must be a string") elif lens and lens not in EXPECTED_LENSES: fail(errors, f"guard references unknown review lens: {lens}") handoff_text = handoff_path.read_text(encoding="utf-8") for required in ( "return_to_router_reason", "Terminal Skills have no outgoing Skill edges", "Every graph cycle must contain an edge with `max_reentries: 1`", ): if required not in handoff_text: fail(errors, f"handoff contract is missing required rule: {required}") router_text = router_path.read_text(encoding="utf-8") if "references/skill-graph.json" not in router_text: fail(errors, "router must reference canonical skill-graph.json") if "skill-graph.yaml" in router_text: fail(errors, "router still references legacy skill-graph.yaml") if "references/handoff-contract.md" not in router_text: fail(errors, "router must reference handoff-contract.md") if "references/agency-role-lenses.md" not in router_text: fail(errors, "router must reference agency-role-lenses.md") skill_texts: dict[str, str] = {} for slug in SPECIALIZED: path = root / "skills" / slug / "SKILL.md" if not path.is_file(): fail(errors, f"missing specialized Skill: {slug}") continue text = path.read_text(encoding="utf-8") skill_texts[slug] = text if "## Connection contract" not in text: fail(errors, f"{slug} is missing a Connection contract section") if "handoff-contract.md" not in text: fail(errors, f"{slug} does not reference handoff-contract.md") if "skill-graph.json" not in text: fail(errors, f"{slug} does not reference skill-graph.json") detector_text = skill_texts.get("ai-writing-detector", "") if "Do not accept a direct handoff from `false-positive-reviewer`" not in detector_text: fail(errors, "ai-writing-detector must explicitly reject direct reviewer handoff") reviewer_text = skill_texts.get("false-positive-reviewer", "") if "This Skill has no direct outgoing Skill edge" not in reviewer_text: fail(errors, "false-positive-reviewer must declare terminal behavior") if "return control to `avoid-ai-writing-router`" not in reviewer_text: fail(errors, "false-positive-reviewer must return new intents to the router") if errors: for error in errors: print(f"ERROR: {error}") print(f"skill connection validation failed with {len(errors)} error(s)") return 1 bounded_edges = sum(1 for edge in edges if isinstance(edge, dict) and edge.get("max_reentries") == 1) print( json.dumps( { "ok": True, "graph_version": graph.get("version"), "skills": len(graph_nodes), "edges": len(edges), "bounded_edges": bounded_edges, "terminal_nodes": sorted(name for name, node in nodes.items() if isinstance(node, dict) and node.get("terminal") is True), "review_lenses": sorted(lenses), }, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())