""" config.py — AIPA Unified Configuration Loads and merges configuration from two sources: 1. config/app.yaml — application structure, paths, infrastructure defaults 2. config/agents.yaml — provider definitions and agent assignments Environment variables override config file values using the AIPA_ prefix. Secrets (API keys) are loaded from .env and never stored in config files. Path resolution order: 1. If AIPA_ROOT is set, relative paths resolve against it. 2. Otherwise, relative paths resolve against the repository root. 3. Absolute paths are always used as-is. Docker pattern: ENV AIPA_ROOT=/app/data VOLUME /app/data # mounts standing_brief, archive, logs, data COPY config/ /app/config/ COPY agents/ /app/agents/ """ import os import warnings from pathlib import Path import yaml from dotenv import load_dotenv # --------------------------------------------------------------------------- # Bootstrap paths (before config is loaded) # --------------------------------------------------------------------------- _REPO_ROOT = Path(__file__).parent.parent load_dotenv(_REPO_ROOT / ".env") # AIPA_ROOT relocates all relative paths — set this in Docker. _AIPA_ROOT = Path(os.getenv("AIPA_ROOT", "")) if os.getenv("AIPA_ROOT") else _REPO_ROOT def _resolve(raw: str) -> Path: """Resolve a path from app.yaml: absolute paths as-is, relative to _AIPA_ROOT.""" p = Path(raw) return p if p.is_absolute() else _AIPA_ROOT / p # --------------------------------------------------------------------------- # Load app.yaml # --------------------------------------------------------------------------- _APP_CONFIG_PATH = Path(os.getenv("AIPA_APP_CONFIG", str(_REPO_ROOT / "config" / "app.yaml"))) if not _APP_CONFIG_PATH.exists(): raise FileNotFoundError( f"Application config not found: {_APP_CONFIG_PATH}\n" f"Expected at config/app.yaml relative to the repo root." ) with open(_APP_CONFIG_PATH, encoding="utf-8") as _f: _app: dict = yaml.safe_load(_f) # --------------------------------------------------------------------------- # Load agents.yaml # --------------------------------------------------------------------------- _AGENTS_CONFIG_PATH = Path( os.getenv("AIPA_AGENTS_CONFIG", str(_resolve(_app["paths"].get("agents_config", "config/agents.yaml")))) ) if not _AGENTS_CONFIG_PATH.exists(): raise FileNotFoundError( f"Agents config not found: {_AGENTS_CONFIG_PATH}\n" f"Expected at config/agents.yaml relative to the repo root." ) with open(_AGENTS_CONFIG_PATH, encoding="utf-8") as _f: _agents_cfg: dict = yaml.safe_load(_f) _providers: dict = _agents_cfg.get("providers", {}) _agents: dict = _agents_cfg.get("agents", {}) # --------------------------------------------------------------------------- # App identity # --------------------------------------------------------------------------- APP_NAME: str = os.getenv("AIPA_APP_NAME", _app["app"]["name"]) APP_VERSION: str = _app["app"]["version"] APP_ENV: str = os.getenv("AIPA_ENV", _app["app"]["environment"]) LOG_LEVEL: str = os.getenv("AIPA_LOG_LEVEL", _app["app"]["log_level"]) # --------------------------------------------------------------------------- # Resolved paths # --------------------------------------------------------------------------- PROMPTS_DIR: Path = _resolve(os.getenv("AIPA_PROMPTS_DIR", _app["paths"]["prompts_dir"])) REGISTRY_PATH: Path = _resolve(os.getenv("AIPA_REGISTRY_PATH", _app["paths"]["registry"])) STANDING_BRIEF_PATH: Path = _resolve(os.getenv("AIPA_STANDING_BRIEF_PATH", _app["paths"]["standing_brief"])) BRIEF_TEMPLATE_PATH: Path = _resolve(os.getenv("AIPA_BRIEF_TEMPLATE_PATH", _app["paths"]["brief_template"])) ARCHIVE_DIR: Path = _resolve(os.getenv("AIPA_ARCHIVE_DIR", _app["paths"]["archive_dir"])) LOGS_DIR: Path = _resolve(os.getenv("AIPA_LOGS_DIR", _app["paths"]["logs_dir"])) DATA_DIR: Path = _resolve(os.getenv("AIPA_DATA_DIR", _app["paths"]["data_dir"])) TASKS_DIR: Path = _resolve(os.getenv("AIPA_TASKS_DIR", _app["paths"].get("tasks_dir", "data/tasks"))) # Derived convenience path — standing brief archive file ARCHIVE_PATH: Path = ARCHIVE_DIR / "standing_brief_archive.md" # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- _log_cfg = _app.get("logging", {}) LOG_FORMAT: str = os.getenv("AIPA_LOG_FORMAT", _log_cfg.get("format", "rich")) _log_file_cfg = _log_cfg.get("file", {}) LOG_FILE_ENABLED: bool = os.getenv("AIPA_LOG_FILE_ENABLED", str(_log_file_cfg.get("enabled", False))).lower() == "true" LOG_FILE_PATH: Path = _resolve(os.getenv("AIPA_LOG_FILE_PATH", _log_file_cfg.get("path", "logs/aipa.log"))) LOG_ROTATION: str = os.getenv("AIPA_LOG_ROTATION", _log_file_cfg.get("rotation", "daily")) LOG_RETENTION_DAYS: int = int(os.getenv("AIPA_LOG_RETENTION_DAYS", str(_log_file_cfg.get("retention_days", 30)))) # --------------------------------------------------------------------------- # Session defaults # --------------------------------------------------------------------------- _sess = _app.get("session", {}) SESSION_LOG_RETENTION: int = int(os.getenv("AIPA_SESSION_LOG_RETENTION", str(_sess.get("log_retention", 10)))) LEAD_TIMEOUT_SECONDS: int = int(os.getenv("AIPA_LEAD_TIMEOUT", str(_sess.get("lead_timeout", 120)))) VERA_AUTO_AUDIT: bool = os.getenv("AIPA_AUTO_AUDIT", str(_sess.get("auto_audit", True))).lower() != "false" DEBUG_PRINT_AGENT_OUTPUTS: bool = os.getenv("AIPA_DEBUG", str(_sess.get("debug_output", False))).lower() == "true" # --------------------------------------------------------------------------- # Memory # --------------------------------------------------------------------------- _mem = _app.get("memory", {}) _mem_store = _mem.get("store", {}) MEMORY_BRIEF_ENABLED: bool = os.getenv("AIPA_MEMORY_BRIEF_ENABLED", str(_mem.get("standing_brief", {}).get("enabled", True))).lower() == "true" MEMORY_STORE_ENABLED: bool = os.getenv("AIPA_MEMORY_STORE_ENABLED", str(_mem_store.get("enabled", False))).lower() == "true" MEMORY_STORE_BACKEND: str = os.getenv("AIPA_MEMORY_STORE_BACKEND", _mem_store.get("backend", "none")) # --------------------------------------------------------------------------- # Infrastructure # --------------------------------------------------------------------------- _infra = _app.get("infrastructure", {}) _api = _infra.get("api", {}) _cache = _infra.get("cache", {}) API_ENABLED: bool = os.getenv("AIPA_API_ENABLED", str(_api.get("enabled", False))).lower() == "true" API_HOST: str = os.getenv("AIPA_API_HOST", _api.get("host", "0.0.0.0")) API_PORT: int = int(os.getenv("AIPA_API_PORT", str(_api.get("port", 8080)))) API_WORKERS: int = int(os.getenv("AIPA_API_WORKERS", str(_api.get("workers", 1)))) CACHE_ENABLED: bool = os.getenv("AIPA_CACHE_ENABLED", str(_cache.get("enabled", False))).lower() == "true" CACHE_BACKEND: str = os.getenv("AIPA_CACHE_BACKEND", _cache.get("backend", "memory")) CACHE_TTL: int = int(os.getenv("AIPA_CACHE_TTL", str(_cache.get("ttl_seconds", 3600)))) # --------------------------------------------------------------------------- # Provider resolution (from agents.yaml) # --------------------------------------------------------------------------- def _provider_def(provider_name: str) -> dict: """Return the provider definition for a named provider.""" if not provider_name or provider_name == "none": return {"type": "none"} pdef = _providers.get(provider_name) if pdef is None: raise ValueError( f"Unknown provider: {provider_name!r}. " f"Define it in config/agents.yaml under 'providers'." ) return pdef def agent_provider_name(agent: str) -> str: """Named provider assigned to an agent (e.g. 'anthropic', 'lmstudio').""" adef = _agents.get(agent, {}) per_agent = adef.get("provider", "") or "" # Treat empty string or the placeholder "none" as "not configured — use default" if per_agent and per_agent != "none": return per_agent return _agents_cfg.get("default_provider", "none") def agent_provider_type(agent: str) -> str: """Provider type: anthropic | openai | openai_compatible | ollama | none.""" return _provider_def(agent_provider_name(agent)).get("type", "none") def agent_api_key(agent: str) -> str: """Resolve the API key for an agent's provider from .env.""" pdef = _provider_def(agent_provider_name(agent)) if "api_key_env" in pdef: key = os.getenv(pdef["api_key_env"], "") if not key: warnings.warn( f"Environment variable {pdef['api_key_env']!r} is not set " f"(required for agent '{agent}'s provider)." ) return key return pdef.get("api_key", "local") def agent_base_url(agent: str) -> str | None: """Resolve the base URL for an agent's provider.""" pname = agent_provider_name(agent) pdef = _provider_def(pname) url = pdef.get("base_url") if not url and pdef.get("type") == "openai_compatible": raise ValueError( f"Provider '{pname}' is type openai_compatible but has no base_url. " f"Add a base_url to its entry in config/agents.yaml." ) return url or None def agent_extra_body(agent: str) -> dict: """Extra request body fields for an agent's provider (e.g. enable_thinking: false).""" pdef = _provider_def(agent_provider_name(agent)) return pdef.get("extra_body") or {} def agent_model(agent: str) -> str: """ Model name for an agent. Uses the agent's own model if set; falls back to the provider's default_model. """ per_agent = _agents.get(agent, {}).get("model", "") or "" if per_agent: return per_agent pdef = _provider_def(agent_provider_name(agent)) return pdef.get("default_model", "") # --------------------------------------------------------------------------- # Role resolution (from agents.yaml `roles` section) # The orchestration code references roles only — never agent names directly. # --------------------------------------------------------------------------- _roles: dict = _agents_cfg.get("roles", {}) ORCHESTRATOR_AGENT: str = _roles.get("orchestrator", "") AUDITOR_AGENT: str = _roles.get("auditor", "") RECRUITER_AGENT: str = _roles.get("recruiter", "") SUMMARIZER_AGENT: str = _roles.get("summarizer", "") if not ORCHESTRATOR_AGENT: raise ValueError("No 'orchestrator' role defined in config/agents.yaml under 'roles'.") if not AUDITOR_AGENT: raise ValueError("No 'auditor' role defined in config/agents.yaml under 'roles'.") # Only the orchestrator, auditor, and summarizer are non-dispatchable. All other # roles (including recruiter) are leads and can receive TASK BRIEFs. _SYSTEM_AGENTS: set[str] = {ORCHESTRATOR_AGENT, AUDITOR_AGENT, SUMMARIZER_AGENT} - {""} # --------------------------------------------------------------------------- # Agent and lead definitions (derived from agents.yaml) # --------------------------------------------------------------------------- def agent_title(agent: str) -> str: """Human-readable title for an agent, from agents.yaml.""" return _agents.get(agent, {}).get("title", agent.capitalize()) def agent_color(agent: str) -> str: """Display color for an agent, from agents.yaml. Defaults to white.""" return _agents.get(agent, {}).get("color", "white") # Full agent config dict used by build_agent() and /agents display. AGENT_CONFIGS: dict[str, dict] = { name: { "prompt_file": adef["prompt_file"], "temperature": adef["temperature"], "max_tokens": adef["max_tokens"], "stateful": adef["stateful"], "tools": list(adef.get("tools") or []), } for name, adef in _agents.items() } def agent_tools(agent: str) -> list[str]: """List of tool names assigned to an agent (empty list if none).""" return AGENT_CONFIGS.get(agent, {}).get("tools", []) def reload_agents_config() -> None: """ Re-read agents.yaml and refresh all derived module-level values. Call after upsert_agent_definition() to pick up changes in the running session. """ global _agents_cfg, _providers, _agents global AGENT_CONFIGS, LEAD_AGENT_NAMES, ACTIVE_PROVIDER global ORCHESTRATOR_AGENT, AUDITOR_AGENT, RECRUITER_AGENT global _roles, _SYSTEM_AGENTS with open(_AGENTS_CONFIG_PATH, encoding="utf-8") as fh: _agents_cfg = yaml.safe_load(fh) _providers = _agents_cfg.get("providers", {}) _agents = _agents_cfg.get("agents", {}) AGENT_CONFIGS = { name: { "prompt_file": adef["prompt_file"], "temperature": adef["temperature"], "max_tokens": adef["max_tokens"], "stateful": adef["stateful"], "tools": list(adef.get("tools") or []), } for name, adef in _agents.items() } _roles = _agents_cfg.get("roles", {}) ORCHESTRATOR_AGENT = _roles.get("orchestrator", "") AUDITOR_AGENT = _roles.get("auditor", "") RECRUITER_AGENT = _roles.get("recruiter", "") SUMMARIZER_AGENT = _roles.get("summarizer", "") _SYSTEM_AGENTS = {ORCHESTRATOR_AGENT, AUDITOR_AGENT, SUMMARIZER_AGENT} - {""} LEAD_AGENT_NAMES = {name for name in _agents if name not in _SYSTEM_AGENTS} ACTIVE_PROVIDER = _agents_cfg.get("default_provider", "none") # Agents the orchestrator can dispatch work to. # Everything in agents.yaml that is not a system-role agent. LEAD_AGENT_NAMES: set[str] = { name for name in _agents if name not in _SYSTEM_AGENTS } # Global fallback provider name (for /status display). ACTIVE_PROVIDER: str = _agents_cfg.get("default_provider", "none") # --------------------------------------------------------------------------- # Knowledge Base # --------------------------------------------------------------------------- _kb = _app.get("knowledge_base", {}) KB_ENABLED: bool = os.getenv("AIPA_KB_ENABLED", str(_kb.get("enabled", True))).lower() == "true" KB_PATH: Path = _resolve(os.getenv("AIPA_KB_PATH", _kb.get("path", "data/knowledge_base"))) KB_DEFAULT_COLLECTION: str = os.getenv("AIPA_KB_DEFAULT_COLLECTION", _kb.get("default_collection", "general")) # --------------------------------------------------------------------------- # Ensure writable runtime directories exist # --------------------------------------------------------------------------- def ensure_runtime_dirs(): """Create runtime directories that may not exist yet (logs, data, archive, tasks).""" for d in (LOGS_DIR, DATA_DIR, ARCHIVE_DIR, TASKS_DIR): d.mkdir(parents=True, exist_ok=True) if KB_ENABLED: KB_PATH.mkdir(parents=True, exist_ok=True)