"""S27-S30: Cross-engine integration checks.""" import json import os import sys from recoil_checks import register_check def _resolve_pipeline_config_path(base): """Locate pipeline_config.json — canonical at recoil/config/pipeline_config.json.""" try: recoil_root = os.path.dirname(os.path.abspath(base)) if recoil_root not in sys.path: sys.path.insert(0, recoil_root) from recoil.core.paths import CONFIG_PATH return str(CONFIG_PATH) except Exception: recoil_root = os.path.dirname(os.path.abspath(base)) return os.path.join(recoil_root, "config", "pipeline_config.json") def check_projects_root_exists(base, discovered): """S27: projects_root from config resolves to an existing directory.""" passes, fails, warns = [], [], [] config_path = _resolve_pipeline_config_path(base) if not os.path.isfile(config_path): warns.append("pipeline_config.json not found") return {"pass": passes, "fail": fails, "warn": warns} with open(config_path) as f: config = json.load(f) projects_root = os.path.expanduser(config.get("projects_root", "")) if not projects_root: fails.append("projects_root not set in config") return {"pass": passes, "fail": fails, "warn": warns} if os.path.isdir(projects_root): passes.append(f"projects_root exists: {projects_root}") # Check default project exists within it default_project = config.get("default_project", "") if default_project: project_dir = os.path.join(projects_root, default_project) if os.path.isdir(project_dir): passes.append(f"Default project '{default_project}' exists in projects_root") else: fails.append(f"Default project '{default_project}' not found at {project_dir}") else: fails.append(f"projects_root does not exist: {projects_root}") return {"pass": passes, "fail": fails, "warn": warns} def check_storyboard_plan_consistency(base, discovered): """S28: Storyboard and plan episode counts are consistent.""" passes, fails, warns = [], [], [] config_path = _resolve_pipeline_config_path(base) if not os.path.isfile(config_path): warns.append("Config not found") return {"pass": passes, "fail": fails, "warn": warns} with open(config_path) as f: config = json.load(f) projects_root = os.path.expanduser(config.get("projects_root", "")) default_project = config.get("default_project", "leviathan") project_dir = os.path.join(projects_root, default_project) if not os.path.isdir(project_dir): warns.append(f"Project directory not found: {project_dir}") return {"pass": passes, "fail": fails, "warn": warns} # Count storyboards storyboards_dir = os.path.join(project_dir, "storyboards") storyboard_count = 0 if os.path.isdir(storyboards_dir): storyboard_count = len([ f for f in os.listdir(storyboards_dir) if f.startswith("storyboard_") and f.endswith(".json") ]) # Count plans plans_dir = os.path.join(base, "data", "plans") plan_count = 0 if os.path.isdir(plans_dir): plan_count = len([ f for f in os.listdir(plans_dir) if f.endswith("_plan.json") ]) passes.append(f"Storyboards: {storyboard_count}, Plans: {plan_count}") if plan_count > 0 and plan_count > storyboard_count: warns.append(f"More plans ({plan_count}) than storyboards ({storyboard_count})") return {"pass": passes, "fail": fails, "warn": warns} def check_recoil_bridge_resolves(base, discovered): """S29: recoil_bridge.py can resolve the default project.""" passes, fails, warns = [], [], [] bridge_path = os.path.join(base, "lib", "recoil_bridge.py") if not os.path.isfile(bridge_path): fails.append("recoil_bridge.py not found") return {"pass": passes, "fail": fails, "warn": warns} passes.append("recoil_bridge.py exists") # Check that projects_root() import is present with open(bridge_path, encoding="utf-8") as f: content = f.read() if "projects_root()" in content: passes.append("projects_root() import present in bridge") else: warns.append("projects_root() not found in bridge (may use RECOIL_ROOT)") if "_project_dir" in content: passes.append("_project_dir() helper defined") else: warns.append("_project_dir() helper not found") return {"pass": passes, "fail": fails, "warn": warns} def check_recoil_engine_accessible(base, discovered): """S30: Recoil engine root is accessible from pipeline config.""" passes, fails, warns = [], [], [] config_path = _resolve_pipeline_config_path(base) if not os.path.isfile(config_path): warns.append("Config not found") return {"pass": passes, "fail": fails, "warn": warns} with open(config_path) as f: config = json.load(f) engine_root = os.path.expanduser(config.get("recoil_engine_root", "")) if not engine_root: warns.append("recoil_engine_root not set in config") return {"pass": passes, "fail": fails, "warn": warns} if os.path.isdir(engine_root): passes.append(f"recoil_engine_root exists: {engine_root}") # Check key engine files key_files = ["CONSTANTS.md", "CLAUDE.md"] key_dirs = ["tools", "lib", "editors"] for f in key_files: if os.path.isfile(os.path.join(engine_root, f)): passes.append(f"Engine file {f} present") else: warns.append(f"Engine file {f} not found") for d in key_dirs: if os.path.isdir(os.path.join(engine_root, d)): passes.append(f"Engine dir {d}/ present") else: warns.append(f"Engine dir {d}/ not found") else: fails.append(f"recoil_engine_root does not exist: {engine_root}") return {"pass": passes, "fail": fails, "warn": warns} register_check("s27_projects_root", "Projects Root Exists", check_projects_root_exists, section="cross", quick=True) register_check("s28_storyboard_plan", "Storyboard ↔ Plan Consistency", check_storyboard_plan_consistency, section="cross") register_check("s29_bridge_resolves", "Recoil Bridge Resolution", check_recoil_bridge_resolves, section="cross", quick=True) register_check("s30_recoil_engine", "Recoil Engine Accessible", check_recoil_engine_accessible, section="cross", quick=True)