import json
import os
from pathlib import Path
from typing import Any, Dict, List
try:
import tomllib
except ImportError:
import tomli as tomllib
class PathValidator:
"""Validates that paths are within the allowed root directory."""
def __init__(self, root_dir: Path):
self.root_dir = root_dir.resolve()
def is_safe(self, path: str | Path) -> bool:
"""Check if a path is safe (within the root directory)."""
try:
# Resolve the path to handle '..' and symlinks
target = Path(path).resolve()
# In Python 3.9+, Path.is_relative_to is the preferred way
return target.is_relative_to(self.root_dir)
except (ValueError, RuntimeError):
return False
class ConfigManager:
"""
Manages Amicus configuration (Global and Workspace).
Priority (highest to lowest):
1. Workspace config (./.amicus/config.toml, ./.amicus/config.json)
2. Global config (~/.config/amicus/config.toml, ~/.config/amicus/config.json)
3. Default values
"""
def __init__(self, workspace_dir: Path = None):
self.workspace_dir = workspace_dir if workspace_dir else Path.cwd()
# Define paths
self.workspace_file_json = self.workspace_dir / ".amicus" / "config.json"
self.workspace_file_toml = self.workspace_dir / ".amicus" / "config.toml"
# Legacy support
if not self.workspace_file_json.exists() and not self.workspace_file_toml.exists() and (self.workspace_dir / "config.json").exists():
self.workspace_file_json = self.workspace_dir / "config.json"
self.global_file_json = Path.home() / ".config" / "amicus" / "config.json"
self.global_file_toml = Path.home() / ".config" / "amicus" / "config.toml"
self.data = self._load()
def _load_file(self, path: Path) -> Dict[str, Any]:
"""Helper to load JSON or TOML file."""
if not path.exists():
return {}
try:
with open(path, "rb") as f:
if path.suffix == ".toml":
return tomllib.load(f)
else:
return json.load(f)
except (ValueError, json.JSONDecodeError, tomllib.TOMLDecodeError):
return {}
def _load(self) -> Dict[str, Any]:
# Default config
config = {
"root_dir": str(self.workspace_dir),
"command_whitelist": ["ls", "git status", "pytest", "grep", "cat", "echo"],
"lock_timeout": 10.0,
"tracking_enabled": False,
"cluster_settings": {
"max_agents": 4,
"idle_timeout_seconds": 30,
"manager_heartbeat_interval": 20,
"workload_assessment_interval": 25,
"grace_period_seconds": 30
},
"model_strengths": {
"gemini-1.5-pro": "high",
"gemini-1.5-flash": "low",
"claude-3-5-sonnet-latest": "high",
"gpt-4o": "high",
"gpt-4o-mini": "low"
},
"roles": {
"bootstrap_manager": {
"preferred_strength": "low",
"description": "Initializes project and manages task distribution."
},
"architect": {
"preferred_strength": "high",
"description": "Designs system structure and complex logic."
},
"developer": {
"preferred_strength": "high",
"description": "Implements features and writes tests."
}
}
}
# Load global config (TOML preferred)
config.update(self._load_file(self.global_file_json))
config.update(self._load_file(self.global_file_toml))
# Load workspace config (TOML preferred)
config.update(self._load_file(self.workspace_file_json))
config.update(self._load_file(self.workspace_file_toml))
# Environment variables override everything
if os.environ.get("AMICUS_LOCK_TIMEOUT"):
try:
config["lock_timeout"] = float(os.environ["AMICUS_LOCK_TIMEOUT"])
except ValueError:
pass
return config
def get(self, key: str, default: Any = None) -> Any:
return self.data.get(key, default)
def save_workspace(self) -> None:
"""Save current config to workspace directory (JSON only for now)."""
# Default to JSON for writing
target = self.workspace_file_json
target.parent.mkdir(parents=True, exist_ok=True)
with open(target, "w") as f:
json.dump(self.data, f, indent=2)
def save_global(self) -> None:
"""Save current config to global directory (JSON only for now)."""
target = self.global_file_json
target.parent.mkdir(parents=True, exist_ok=True)
with open(target, "w") as f:
json.dump(self.data, f, indent=2)
