diff options
Diffstat (limited to 'scripts/workflow-integrity.py')
| -rwxr-xr-x | scripts/workflow-integrity.py | 158 |
1 files changed, 158 insertions, 0 deletions
diff --git a/scripts/workflow-integrity.py b/scripts/workflow-integrity.py new file mode 100755 index 0000000..fa33c4c --- /dev/null +++ b/scripts/workflow-integrity.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +"""Integrity checks for the .ai/workflows/ directory and its INDEX.org. + +Startup's drift check catches index-vs-directory mismatches; this goes deeper: +a workflow referencing a renamed script, a plugin whose engine was deleted, a +missing required section, a duplicate trigger phrase. Runs against the canonical +claude-templates/.ai/workflows/ by default; pass a directory to check another. + +Checks: + 1. indexed-or-plugin every *.org (except INDEX.org) is a catalog entry in + INDEX.org, or a source plugin (engine.plugin.org) of an + indexed engine. + 2. indexed-exists every catalog entry points at a file that exists. + 3. script-refs every .ai/scripts/<x> reference in a workflow resolves + to a real file under the canonical scripts dir. + 4. plugin-parent every engine.plugin.org maps to an indexed engine. + 5. orientation every non-plugin workflow has an orientation section + (Overview / Purpose / When to Use|Run / Status). + 6. trigger-uniqueness no trigger phrase is claimed by two different workflows. + +Exit 0 when clean, 1 when any check fails, 2 on bad usage. + +Usage: + workflow-integrity.py [WORKFLOWS_DIR] +""" +from __future__ import annotations + +import re +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +DEFAULT_DIR = REPO / "claude-templates" / ".ai" / "workflows" + +CATALOG_RE = re.compile(r"^- =([A-Za-z0-9._-]+\.org)= ", re.M) +PLUGIN_RE = re.compile(r"^(.+)\.[^.]+\.org$") # engine.plugin.org +SCRIPT_REF_RE = re.compile(r"\.ai/scripts/([A-Za-z0-9][A-Za-z0-9._/-]*)") +QUOTED_RE = re.compile(r'"([^"]+)"') +ORIENTATION_RE = re.compile(r"^\* (overview|purpose|when to use|when to run|status)\b", + re.M | re.I) + + +def catalog_entries(index_text: str) -> set[str]: + """Indexed workflow filenames — list items of the form `- =name.org= — ...`. + Distinguishes catalog entries from prose mentions like =todo.org=.""" + return set(CATALOG_RE.findall(index_text)) + + +def scripts_dir_for(workflows_dir: Path) -> Path: + """The scripts dir a workflows dir's .ai/scripts/ references resolve against.""" + return workflows_dir.parent / "scripts" + + +def trigger_map(index_text: str) -> dict[str, set[str]]: + """Map each trigger phrase to the set of workflows that claim it. + + Tracks the current workflow as catalog entries are seen, then attributes + quoted phrases on any subsequent 'trigger' line to it. + """ + phrases: dict[str, set[str]] = {} + current: str | None = None + for line in index_text.splitlines(): + m = CATALOG_RE.match(line) + if m: + current = m.group(1) + continue + if current and "rigger" in line: # "Triggers:", "Full-prep triggers:", etc. + for phrase in QUOTED_RE.findall(line): + phrases.setdefault(phrase, set()).add(current) + return phrases + + +def check(workflows_dir: Path) -> list[str]: + findings: list[str] = [] + index = workflows_dir / "INDEX.org" + if not index.is_file(): + return [f"no INDEX.org in {workflows_dir}"] + index_text = index.read_text(encoding="utf-8") + indexed = catalog_entries(index_text) + engines = {n[:-len(".org")] for n in indexed} + workflows = sorted(p.name for p in workflows_dir.glob("*.org") if p.name != "INDEX.org") + scripts_dir = scripts_dir_for(workflows_dir) + + def is_plugin(name: str) -> bool: + return any(name.startswith(e + ".") and name != e + ".org" for e in engines) + + # 1. indexed-or-plugin + for w in workflows: + if w not in indexed and not is_plugin(w): + findings.append(f"[orphan] {w}: not indexed in INDEX.org and not a plugin of an indexed engine") + + # 2. indexed-exists + for n in sorted(indexed): + if not (workflows_dir / n).is_file(): + findings.append(f"[stale-index] INDEX lists {n} but no such file exists") + + # 4. plugin-parent + for w in workflows: + if w in indexed: + continue + m = PLUGIN_RE.match(w) + if m: + parent = m.group(1) + ".org" + if parent not in indexed: + findings.append(f"[orphan-plugin] {w}: parent engine {parent} is not indexed") + elif not (workflows_dir / parent).is_file(): + findings.append(f"[orphan-plugin] {w}: parent engine {parent} is missing") + + # 3. script-refs + for w in workflows: + text = (workflows_dir / w).read_text(encoding="utf-8") + for ref in sorted(set(SCRIPT_REF_RE.findall(text))): + ref = ref.rstrip(".") + if (scripts_dir / ref).exists(): + continue + if (scripts_dir / ref.split("/")[0]).exists(): + continue + findings.append(f"[bad-ref] {w}: references .ai/scripts/{ref} which does not exist") + + # 5. orientation (plugins are adapters loaded by their engine — exempt) + for w in workflows: + if is_plugin(w): + continue + text = (workflows_dir / w).read_text(encoding="utf-8") + if not ORIENTATION_RE.search(text): + findings.append(f"[missing-section] {w}: no orientation section " + "(Overview / Purpose / When to Use|Run / Status)") + + # 6. trigger-uniqueness + for phrase, owners in sorted(trigger_map(index_text).items()): + if len(owners) > 1: + findings.append(f"[dup-trigger] \"{phrase}\" claimed by {', '.join(sorted(owners))}") + + return findings + + +def main() -> int: + if len(sys.argv) > 2: + print("usage: workflow-integrity.py [WORKFLOWS_DIR]", file=sys.stderr) + return 2 + workflows_dir = Path(sys.argv[1]) if len(sys.argv) == 2 else DEFAULT_DIR + if not workflows_dir.is_dir(): + print(f"workflow-integrity: {workflows_dir} is not a directory", file=sys.stderr) + return 2 + + findings = check(workflows_dir) + n_workflows = len([p for p in workflows_dir.glob("*.org") if p.name != "INDEX.org"]) + if findings: + print("workflow-integrity: FAIL") + for f in findings: + print(f" {f}") + return 1 + print(f"workflow-integrity: OK ({n_workflows} workflows)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) |