""" DSS Context Compiler. Resolves project context via 3-layer cascade: Base -> Skin -> Project Includes Safe Boot Protocol and Debug Provenance. """ import copy import json import logging from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("DSSCompiler") # --- SAFE BOOT PROTOCOL --- # Hardcoded emergency skin in case file system or JSON parsing fails catastrophicly EMERGENCY_SKIN = { "meta": {"id": "emergency", "version": "1.0.0"}, "tokens": { "colors": {"primary": "#FF0000", "background": "#FFFFFF", "text": "#000000"}, "spacing": {"base": "4px"}, }, "status": "emergency_mode", } class ContextCompiler: def __init__(self, skins_dir: str = "./skins"): self.skins_dir = Path(skins_dir) self.cache: Dict[str, Any] = {} self._manifest_mtimes: Dict[str, float] = {} # Track file modification times def compile( self, manifest_path: str, debug: bool = False, force_refresh: bool = False ) -> Dict[str, Any]: """ Main entry point. Compiles context by merging: 1. Base Skin (Implicit or Explicit) 2. Extended Skin (defined in manifest) 3. Project Overrides (defined in manifest) Args: manifest_path: Path to ds.config.json debug: Enable provenance tracking force_refresh: Bypass cache and recompile (for long-running servers) """ try: # Check cache with mtime validation (unless force_refresh or debug mode) # Note: Debug mode bypasses cache because provenance must be recalculated cache_key = f"{manifest_path}:debug={debug}" if not force_refresh and not debug and cache_key in self.cache: # Verify manifest hasn't changed manifest_file = Path(manifest_path) if manifest_file.exists(): current_mtime = manifest_file.stat().st_mtime cached_mtime = self._manifest_mtimes.get(cache_key, 0) if current_mtime == cached_mtime: logger.debug(f"Cache hit for {manifest_path}") return self.cache[cache_key] else: logger.info(f"Manifest modified, invalidating cache: {manifest_path}") # 1. Load Project Manifest manifest = self._load_json(manifest_path) # 2. Resolve Skin skin_id = manifest.get("extends", {}).get("skin", "classic") skin = self._load_skin(skin_id) # 3. Resolve Base (Single Inheritance Enforced) # If the skin extends another, we merge that first. # Simplified for Phase 1: We assume all skins extend 'base' implicitly unless specified base_skin = self._load_skin("base") # 4. Cascade Merge: Base -> Skin -> Project # Merge Base + Skin context = self._deep_merge(base_skin, skin, path="base->skin", debug=debug) # Merge Result + Project Overrides # Need to wrap project overrides in same structure as skins project_overrides_wrapped = {"tokens": manifest.get("overrides", {}).get("tokens", {})} final_context = self._deep_merge( context, project_overrides_wrapped, path="skin->project", debug=debug ) # Inject Metadata final_context["_meta"] = { "project_id": manifest["project"]["id"], "compiled_at": datetime.now(timezone.utc).isoformat(), "debug_enabled": debug, "compiler_config": manifest.get("compiler", {}), } if debug: final_context["_provenance"] = self.provenance_log # Cache result with mtime tracking (only cache non-debug mode results) if not debug: manifest_file = Path(manifest_path) if manifest_file.exists(): cache_key = f"{manifest_path}:debug={debug}" self.cache[cache_key] = final_context self._manifest_mtimes[cache_key] = manifest_file.stat().st_mtime logger.debug(f"Cached compilation result for {manifest_path}") return final_context except Exception as e: logger.error(f"Compiler specific error: {e}") logger.warning("Initiating SAFE BOOT PROTOCOL") return self._enter_safe_mode(e) def _load_skin(self, skin_id: str) -> Dict[str, Any]: """Loads a skin by ID from the skins directory.""" # Simple caching strategy if skin_id in self.cache: return self.cache[skin_id] # Security: Prevent path traversal attacks path = (self.skins_dir / f"{skin_id}.json").resolve() if not str(path).startswith(str(self.skins_dir.resolve())): raise ValueError(f"Invalid skin ID (path traversal detected): {skin_id}") if not path.exists(): logger.warning(f"Skin {skin_id} not found, falling back to base.") if skin_id == "base": # Return emergency tokens if base is missing return EMERGENCY_SKIN return self._load_skin("base") data = self._load_json(str(path)) self.cache[skin_id] = data return data def _load_json(self, path: str) -> Dict[str, Any]: with open(path, "r") as f: return json.load(f) def _deep_merge( self, base: Dict, override: Dict, path: str = "", debug: bool = False, provenance: List[Dict] = None, ) -> Dict: """ Deep merge dictionaries. Replaces arrays. Populates provenance list if debug is True (thread-safe). """ # Thread-safe: use method parameter instead of instance variable if provenance is None and debug: provenance = [] # Store reference on first call for later retrieval if not hasattr(self, "provenance_log"): self.provenance_log = provenance result = copy.deepcopy(base) for key, value in override.items(): if isinstance(value, dict) and key in result and isinstance(result[key], dict): # Recursive merge - pass provenance down result[key] = self._deep_merge( result[key], value, path=f"{path}.{key}", debug=debug, provenance=provenance ) else: # Direct replacement (Primitive or Array) if debug and provenance is not None: provenance.append( { "key": key, "action": "override", "layer": path, "value_type": type(value).__name__, } ) result[key] = copy.deepcopy(value) return result def _enter_safe_mode(self, error: Exception) -> Dict[str, Any]: """Returns the hardcoded emergency skin with error details.""" safe_context = copy.deepcopy(EMERGENCY_SKIN) safe_context["_error"] = str(error) return safe_context