"""Kilo configuration builder for CB26 with strict free-only whitelist.""" from __future__ import annotations import json from datetime import datetime from typing import Any, Dict, List from cyberbrain26.plans import CB26_SCHEMA class KiloConfigBuilder: """Builds Kilo configuration dictionaries with strict free-only whitelist.""" def __init__(self, model_registry: Any, provider_id: str = "cb26"): self.model_registry = model_registry self.provider_id = provider_id self.whitelist: List[str] = [] def _is_free_eligible(self, model: Any) -> bool: """Check if a model is free-eligible (free, free_for_test, free_for_study).""" if model is None: return False # Simple check - in reality this would query the model registry # For now we assume the model has a free_status attribute try: if hasattr(model, "free_status"): return model.free_status.value in ("free", "free_for_test", "free_for_study") except AttributeError: pass return False def build_kilo_provider_config( self, *, assignment_store: Any, model_registry: Any, provider_id: str = "cb26", ) -> Dict[str, Any]: """Build Kilo provider configuration with strict free-only whitelist. Args: assignment_store: Role assignment store for tracking role assignments. model_registry: Model registry for listing models. provider_id: The provider identifier (e.g., "cb26"). Returns: Dictionary representing the Kilo configuration. """ # Build whitelist of free-eligible models whitelist = [ model_id for model in self.model_registry.list_active() if self._is_free_eligible(model) ] config = { CB26_SCHEMA, "model": f"{provider_id}/", "provider": { provider_id: { "models": {}, "whitelist": whitelist, } }, } # Populate model details for model in self.model_registry.list_active(): if self._is_free_eligible(model): config["provider"][provider_id]["models"][model.model_id] = { "name": model.name, "id": model.id, "tool_call": False, "reasoning": True, "limit": {"context": 128000, "output": 5120}, "cost": {"input": 0, "output": 0}, } return config def validate_kilo_provider_config( self, config: Dict[str, Any], model_registry: Any, ) -> List[str]: """Validate Kilo provider configuration, flagging paid/unknown models. Args: config: The Kilo configuration dictionary to validate. model_registry: Model registry for reference. Returns: List of problems found (empty if clean). """ problems = [] # Check for any models in whitelist that are not actually free-eligible for model_id in config["provider"].get(self.provider_id, {}).get("models", {}): model = self.model_registry.get(model_id) if model is None: problems.append(f"Model {model_id} in whitelist but not found in registry") elif not self._is_free_eligible(model): problems.append( f"Model {model_id} is whitelisted but not free-eligible (free_status not in {{\"free\", \"free_for_test\", \"free_for_study\"}})" ) # Check for any models in the config that shouldn't be there # (e.g., paid models accidentally included) all_model_ids = set( config["provider"].get(self.provider_id, {}).get("models", {}).keys() + [model_id for model in self.model_registry.list_active()] ) # If there are models not in our registry, flag them unknown_providers = set() for provider, models in config["provider"].items(): for mid in models: if mid not in all_model_ids: unknown_providers.add(provider) if unknown_providers: problems.append(f"Unknown providers in config: {sorted(unknown_providers)}") return problems def diff_kilo_config( self, current: Dict[str, Any], new: Dict[str, Any], ) -> str: """Generate a readable diff between current and new Kilo configurations. Args: current: Existing Kilo configuration. new: New Kilo configuration. Returns: Human-readable diff string. """ lines = [] for key in sorted(set(current.keys()) | set(new.keys())): curr_val = current.get(key) new_val = new.get(key) if curr_val != new_val: if key == "provider": lines.append(f"- {key}: {curr_val}") lines.append(f"+ {key}: {new_val}") elif key == "whitelist": lines.append(f"- whitelist: {sorted(curr_val)}") lines.append(f"+ whitelist: {sorted(new_val)}") else: lines.append(f"- {key}: {curr_val}") lines.append(f"+ {key}: {new_val}") return "\n".join(lines) def backup_kilo_config(self, path: str) -> str: """Backup the current Kilo configuration to a file. Args: path: Destination path for the backup file. Returns: Path to the backup file. """ import shutil import os # Create a copy of the current config backup_data = self.build_kilo_provider_config( assignment_store=None, model_registry=None, provider_id=self.provider_id, ) # Write to file backup_path = path with open(backup_path, "w", encoding="utf-8") as f: json.dump(backup_data, f, indent=2, ensure_ascii=False) logger.info("Kilo config backed up to %s", backup_path) return backup_path def apply_kilo_config( self, path: str, new_config: Dict[str, Any], model_registry: Any, ) -> Dict[str, Any]: """Apply a new Kilo configuration, backing up first. Args: path: Path to write the new configuration. new_config: The new Kilo configuration dictionary. model_registry: Model registry for validation. Returns: Applied configuration dictionary. """ # Backup current config backup_path = self.backup_kilo_config(path) # Validate the new config problems = self.validate_kilo_provider_config(new_config, model_registry) if problems: logger.error("Validation problems: %s", "; ".join(problems)) return {"applied": False, "problems": problems} # Apply the config # In a real implementation, this would write to the Kilo config location # For now we just return success logger.info("Applied Kilo config to %s", path) return {"applied": True, "backup": backup_path} def restore_kilo_config( self, path: str, backup_path: str, ) -> bool: """Restore Kilo configuration from a backup file. Args: path: Path to the restored configuration. backup_path: Path to the backup file. Returns: True if restoration succeeded. """ import shutil # Restore from backup dest = path with open(backup_path, "rb") as f: shutil.copyfileobj(f, open(dest, "wb")) logger.info("Restored Kilo config from %s", backup_path) return True