#!/usr/bin/env python3 """Dual-publish drift guard (audit gate G4). Several skills are published twice on purpose: bundled: /skills//... standalone: //skills//... or /compliance-team-*/skills//... (ra-qm-team pattern) The two copies are kept in sync by scripts/sync_skill_bundles.py. This guard discovers every such pair programmatically and recursively compares the two trees (a `diff -rq` equivalent built on os.walk + filecmp with full content comparison). Any file that exists on only one side, or whose content differs, is drift. Exit codes: 0 all pairs identical 1 drift detected (drifted files listed on stdout) 2 unexpected error (e.g. repo root not found) Usage: python3 scripts/check_dual_publish.py # check all pairs python3 scripts/check_dual_publish.py --list # show discovered pairs python3 scripts/check_dual_publish.py --json # machine-readable report """ from __future__ import annotations import argparse import filecmp import json import os import sys REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Top-level directories that are never skill domains (sync copies, docs, VCS). EXCLUDE_TOP_LEVEL = { ".git", ".github", ".codex", ".gemini", ".hermes", ".vibe", ".claude", ".claude-plugin", ".codex-plugin", "docs", "audit", "node_modules", "integrations", "eval-workspace", "templates", "standards", "assets", "scripts", "agents", "commands", "orchestration", "custom-gpt", "site", } # Directory names that may never appear inside a compared payload # (none expected today; kept for future-proofing against editor litter). IGNORE_NAMES = {"__pycache__", ".DS_Store"} def is_skill_dir(path): return os.path.isfile(os.path.join(path, "SKILL.md")) def discover_pairs(repo_root): """Return [(bundled_rel, standalone_rel)] for every dual-publish pair.""" pairs = [] for domain in sorted(os.listdir(repo_root)): domain_path = os.path.join(repo_root, domain) if domain in EXCLUDE_TOP_LEVEL or domain.startswith("."): continue if not os.path.isdir(domain_path): continue bundled_root = os.path.join(domain_path, "skills") if not os.path.isdir(bundled_root): continue for name in sorted(os.listdir(bundled_root)): bundled = os.path.join(bundled_root, name) if not is_skill_dir(bundled): continue candidates = [os.path.join(domain_path, name, "skills", name)] for entry in sorted(os.listdir(domain_path)): if entry.startswith("compliance-team-"): candidates.append( os.path.join(domain_path, entry, "skills", name)) for cand in candidates: if is_skill_dir(cand): pairs.append(( os.path.relpath(bundled, repo_root), os.path.relpath(cand, repo_root), )) return pairs def _listdir(path): try: return sorted(e for e in os.listdir(path) if e not in IGNORE_NAMES) except OSError: return [] def compare_trees(left, right, rel=""): """Recursive diff -rq equivalent. Returns list of drift descriptions.""" drift = [] left_entries = set(_listdir(left)) right_entries = set(_listdir(right)) for entry in sorted(left_entries - right_entries): drift.append(f"only-in-bundled: {os.path.join(rel, entry)}") for entry in sorted(right_entries - left_entries): drift.append(f"only-in-standalone: {os.path.join(rel, entry)}") for entry in sorted(left_entries & right_entries): lp = os.path.join(left, entry) rp = os.path.join(right, entry) sub_rel = os.path.join(rel, entry) l_dir, r_dir = os.path.isdir(lp), os.path.isdir(rp) if l_dir != r_dir: drift.append(f"type-mismatch (file vs dir): {sub_rel}") elif l_dir: drift.extend(compare_trees(lp, rp, sub_rel)) else: try: same = filecmp.cmp(lp, rp, shallow=False) except OSError as exc: drift.append(f"unreadable: {sub_rel} ({exc})") continue if not same: drift.append(f"differs: {sub_rel}") return drift def main(argv=None): parser = argparse.ArgumentParser( description="Verify dual-published skill pairs are byte-identical " "(gate G4).") parser.add_argument("--list", action="store_true", help="list discovered pairs and exit") parser.add_argument("--json", action="store_true", help="emit a JSON report instead of human-readable output") parser.add_argument("--root", default=REPO_ROOT, help="repo root (default: parent of this script)") args = parser.parse_args(argv) if not os.path.isdir(args.root): print(f"ERROR: repo root not found: {args.root}", file=sys.stderr) return 2 pairs = discover_pairs(args.root) if args.list: if args.json: print(json.dumps( [{"bundled": b, "standalone": s} for b, s in pairs], indent=2)) else: for bundled, standalone in pairs: print(f"{bundled} <-> {standalone}") print(f"\n{len(pairs)} dual-publish pair(s) discovered") return 0 report = [] drifted_pairs = 0 for bundled, standalone in pairs: drift = compare_trees( os.path.join(args.root, bundled), os.path.join(args.root, standalone)) report.append({"bundled": bundled, "standalone": standalone, "drift": drift}) if drift: drifted_pairs += 1 if args.json: print(json.dumps({ "pairs": len(pairs), "drifted": drifted_pairs, "results": report, }, indent=2)) else: for item in report: status = "DRIFT" if item["drift"] else "OK" print(f"[{status}] {item['bundled']} <-> {item['standalone']}") for line in item["drift"]: print(f" {line}") print(f"\n{len(pairs)} pair(s) checked, {drifted_pairs} drifted") if not pairs: print("WARNING: no dual-publish pairs discovered — discovery logic " "may be stale", file=sys.stderr) return 1 if drifted_pairs else 0 if __name__ == "__main__": sys.exit(main())