#!/usr/bin/env python3 """ Engine protocol integrity checker. Modular audit system — discovers files, runs registered checks, reports results. Check modules live in engine_checks/ and self-register on import. 53 total checks across 8 categories: Structural (1-6): chain, gates, constants, cross-refs, commands, orphans Semantic (7-18): stale values, counting, hooks, templates, examples, etc. Deep Logic (19-23): JSON handling, thresholds, state fields, runtime, prose refs GUI Integrity (24-30): routes, onclick, modules, standalone, launch, server, secrets Visual Pipeline (31-38): config loader, engine cost map, pricing, project config, etc. Dead Code (39-42): unreferenced tools, orphaned editors, unused imports, archive dupes Documentation (43-49): CLAUDE.md tables, folder structure, cost figures, workflow guide Security (50-53): API keys, env vars, path traversal, config secrets Usage: python3 engine_check.py /path/to/recoil python3 engine_check.py /path/to/recoil --json python3 engine_check.py /path/to/recoil --verbose python3 engine_check.py /path/to/recoil --section chain python3 engine_check.py /path/to/recoil --section security python3 engine_check.py /path/to/recoil --section gui python3 engine_check.py /path/to/recoil --quick python3 engine_check.py /path/to/recoil --fix Exit codes: 0 = all checks pass 1 = failures found 2 = script error """ import argparse import json import os import sys # ═══════════════════════════════════════════════════════════════ # DIRECTORY SCANNING — discovers files automatically # ═══════════════════════════════════════════════════════════════ SCAN_DIRS = [ "tools", "lib", "agents", "editors", "templates", "schemas", "skills", "lenses", "evaluation", "calibration", "docs", "config", ".claude/hooks", ".claude/skills", ] SCAN_EXTENSIONS = {".md", ".py", ".json", ".html"} EXCLUDE_PATTERNS = {".DS_Store", "__pycache__", "docs/archive", ".venv", "node_modules"} def discover_files(base): """Scan project directories, return {relative_path: True} for all files.""" files = {} for name in ["CLAUDE.md"]: if os.path.exists(os.path.join(base, name)): files[name] = True for scan_dir in SCAN_DIRS: dir_path = os.path.join(base, scan_dir) if not os.path.isdir(dir_path): continue for root, dirs, filenames in os.walk(dir_path): dirs[:] = [d for d in dirs if d not in EXCLUDE_PATTERNS and not any(os.path.join(root, d).endswith(ep) for ep in EXCLUDE_PATTERNS)] for filename in filenames: if filename.startswith("."): continue if os.path.splitext(filename)[1] not in SCAN_EXTENSIONS: continue full = os.path.join(root, filename) rel = os.path.relpath(full, base).replace(os.sep, "/") if not any(excl in rel for excl in EXCLUDE_PATTERNS): files[rel] = True return files # ═══════════════════════════════════════════════════════════════ # IMPORT CHECK REGISTRY # ═══════════════════════════════════════════════════════════════ # Add this directory to path so engine_checks package can be imported _tools_dir = os.path.dirname(os.path.abspath(__file__)) if _tools_dir not in sys.path: sys.path.insert(0, _tools_dir) from engine_checks import ( get_all_checks, get_section_map, get_quick_keys, get_fixes_for_keys, ) # ═══════════════════════════════════════════════════════════════ # RUN AND REPORT # ═══════════════════════════════════════════════════════════════ def run_checks(base, section=None, quick=False): """Discover files, then run protocol integrity checks.""" discovered = discover_files(base) all_checks = get_all_checks() section_map = get_section_map() quick_keys = get_quick_keys() checks_to_run = all_checks if quick: checks_to_run = [(k, n, fn) for k, n, fn in all_checks if k in quick_keys] elif section and section in section_map: keys = set(section_map[section]) checks_to_run = [(k, n, fn) for k, n, fn in all_checks if k in keys] sections = {} total_pass = total_fail = total_warn = 0 for key, name, fn in checks_to_run: result = fn(base, discovered) sections[key] = {"name": name, **result} total_pass += len(result.get("pass", [])) total_fail += len(result.get("fail", [])) total_warn += len(result.get("warn", [])) return { "discovered_files": len(discovered), "sections": sections, "summary": { "total_pass": total_pass, "total_fail": total_fail, "total_warn": total_warn, "overall": "PASS" if total_fail == 0 else "FAIL", }, } def run_fixes(base, results): """Apply registered auto-fixes for checks that have warnings or failures. Returns dict with fix results: {"total_fixed": N, "total_skipped": N, "details": [...]} """ discovered = discover_files(base) failed_or_warned_keys = [ key for key, section in results["sections"].items() if section.get("fail") or section.get("warn") ] available_fixes = get_fixes_for_keys(failed_or_warned_keys) if not available_fixes: return {"total_fixed": 0, "total_skipped": 0, "details": []} all_fixed = [] all_skipped = [] for key, fix_fn in available_fixes.items(): try: fix_result = fix_fn(base, discovered) fixed = fix_result.get("fixed", []) skipped = fix_result.get("skipped", []) all_fixed.extend(fixed) all_skipped.extend(skipped) except Exception as e: all_skipped.append(f"{key}: fix error — {e}") return { "total_fixed": len(all_fixed), "total_skipped": len(all_skipped), "details": all_fixed + [f"SKIPPED: {s}" for s in all_skipped], } def format_report(results, verbose=False): """Format human-readable report.""" lines = [] lines.append("=" * 65) lines.append("ENGINE PROTOCOL INTEGRITY CHECK") lines.append("=" * 65) lines.append("") s = results["summary"] lines.append(f"RESULT: {s['overall']}") lines.append(f" Scanned: {results['discovered_files']} files") lines.append(f" Pass: {s['total_pass']}") lines.append(f" Fail: {s['total_fail']}") lines.append(f" Warn: {s['total_warn']}") lines.append("") for _key, section in results["sections"].items(): name = section["name"] passes = section.get("pass", []) fails = section.get("fail", []) warns = section.get("warn", []) icon = "PASS" if not fails else "FAIL" lines.append( f"[{icon}] {name}: " f"{len(passes)} pass, {len(fails)} fail, {len(warns)} warn" ) for f in fails: lines.append(f" FAIL {f}") for w in warns: lines.append(f" WARN {w}") if verbose: for p in passes: lines.append(f" OK {p}") lines.append("") lines.append("=" * 65) lines.append("") lines.append("Check modules: engine_checks/ (structural, semantic, deep_logic,") lines.append("gui_integrity, visual_pipeline, dead_code, documentation, security)") lines.append("Add new checks by creating a module and calling register_check().") return "\n".join(lines) def main(): parser = argparse.ArgumentParser( description="Engine protocol integrity checker" ) parser.add_argument("base_path", help="Path to Recoil root directory") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--verbose", action="store_true", help="Show passing checks too") parser.add_argument( "--section", choices=sorted(get_section_map().keys()), help="Run only one check section", ) parser.add_argument( "--quick", action="store_true", help="Run only quick checks (chain, inventory, json, python, security)", ) parser.add_argument( "--full", action="store_true", help="Run all checks (default)", ) parser.add_argument( "--fix", action="store_true", help="Apply safe auto-fixes (review with git diff, commit manually)", ) args = parser.parse_args() if not os.path.isdir(args.base_path): print(f"ERROR: Not a directory: {args.base_path}") sys.exit(2) try: results = run_checks(args.base_path, section=args.section, quick=args.quick) except Exception as e: print(f"ERROR: {e}") import traceback traceback.print_exc() sys.exit(2) if args.fix: fix_results = run_fixes(args.base_path, results) if fix_results["total_fixed"] > 0: # Re-run checks after fixes results = run_checks(args.base_path, section=args.section, quick=args.quick) if args.json: results["fixes"] = fix_results print(json.dumps(results, indent=2)) else: print(format_report(results, verbose=args.verbose)) if fix_results["total_fixed"] > 0 or fix_results["total_skipped"] > 0: print("") print("=" * 65) print("AUTO-FIXES APPLIED") print("=" * 65) for detail in fix_results["details"]: print(f" {detail}") print(f"\n Fixed: {fix_results['total_fixed']}") print(f" Skipped: {fix_results['total_skipped']}") print("\n Review changes: git diff") print(" Commit when ready: git add -A && git commit") else: print("\n No auto-fixes available for current issues.") elif args.json: print(json.dumps(results, indent=2)) else: print(format_report(results, verbose=args.verbose)) sys.exit(0 if results["summary"]["overall"] == "PASS" else 1) if __name__ == "__main__": main()