#!/usr/bin/env python3 """ Pipeline engine integrity checker. Modular audit system — mirrors recoil/tools/engine_check.py pattern. Check modules live in recoil_checks/ and self-register on import. 30 total checks across 6 categories: API & Config (S1-S4): model_profiles, pipeline_config, env vars, model refs Execution Store (S5-S8): DB schema, orphaned shots, takes JSON, WAL mode Review Server (S9-S14): routes, endpoints, static paths, HTML, port, CORS Console Frontend (S15-S20): tab JS, console app, onclick, morphdom, XSS, tabs Pipeline (S21-S26): sub-pipelines, costs, templates, schema, manifests, validation Cross-Engine (S27-S30): projects_root, storyboard/manifest, bridge, recoil engine Usage: python3 recoil_check.py /path/to/recoil/pipeline python3 recoil_check.py /path/to/recoil/pipeline --json python3 recoil_check.py /path/to/recoil/pipeline --verbose python3 recoil_check.py /path/to/recoil/pipeline --section api python3 recoil_check.py /path/to/recoil/pipeline --quick Exit codes: 0 = all checks pass 1 = failures found 2 = script error """ import argparse import json import os import sys # ═══════════════════════════════════════════════════════════════ # DIRECTORY SCANNING # ═══════════════════════════════════════════════════════════════ SCAN_DIRS = [ "lib", "orchestrator", "tools", "editors", "config", "tests", ] SCAN_EXTENSIONS = {".md", ".py", ".json", ".html", ".js"} EXCLUDE_PATTERNS = {".DS_Store", "__pycache__", ".venv", "node_modules", ".pytest_cache"} def discover_files(base): """Scan directories, return {relative_path: True} for all files.""" files = {} for name in ["CLAUDE.md", ".gitignore"]: 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] 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 # ═══════════════════════════════════════════════════════════════ _tools_dir = os.path.dirname(os.path.abspath(__file__)) if _tools_dir not in sys.path: sys.path.insert(0, _tools_dir) from recoil_checks import ( get_all_checks, get_section_map, get_quick_keys, ) # ═══════════════════════════════════════════════════════════════ # RUN AND REPORT # ═══════════════════════════════════════════════════════════════ def run_checks(base, section=None, quick=False): """Discover files, run checks, return results.""" 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: try: result = fn(base, discovered) except Exception as e: result = {"pass": [], "fail": [f"Check crashed: {e}"], "warn": []} 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 format_report(results, verbose=False): """Format human-readable report.""" lines = [] lines.append("=" * 65) lines.append("PIPELINE ENGINE 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: recoil_checks/ (api_config, execution_store,") lines.append("review_server, console_frontend, pipeline_integrity, cross_engine)") return "\n".join(lines) def main(): parser = argparse.ArgumentParser( description="Starsend engine integrity checker" ) parser.add_argument("base_path", help="Path to Starsend root directory") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--verbose", action="store_true", help="Show passing checks") 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 quick checks only") 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.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()