"""Actor loader for discovering and caching actor YAML configurations. Discovers actor YAML files from configurable search roots, parses them against the v3 ``ActorConfigSchema``, applies namespace defaults, and caches results using content hashing to avoid redundant reloads. Operations: | Method | Description | |--------------------------|------------------------------------------------| | ``discover()`` | Scan search roots and load/reload actors | | ``get(name)`` | Retrieve a loaded actor by namespaced name | | ``list_actors(...)`` | List actors with optional namespace filter | | ``clear()`` | Drop all cached actors and hashes | All mutating operations are protected by ``threading.RLock``. """ from __future__ import annotations import hashlib import logging import threading from pathlib import Path from typing import Any import yaml from cleveragents.actor.schema import ActorConfigSchema from cleveragents.core.exceptions import ValidationError from cleveragents.tool.registry import ToolRegistry logger = logging.getLogger(__name__) _DEFAULT_NAMESPACE = "local" _YAML_SUFFIXES = frozenset({".yaml", ".yml"}) class _CacheEntry: """Internal cache entry storing an actor config alongside its content hash.""" __slots__ = ("config", "content_hash", "load_count", "source_path") def __init__( self, config: ActorConfigSchema, content_hash: str, source_path: Path, ) -> None: self.config = config self.content_hash = content_hash self.source_path = source_path self.load_count = 1 def _compute_hash(content: bytes) -> str: return hashlib.sha256(content).hexdigest() def _normalize_name(raw_name: str) -> str: """Apply ``local/`` default when the name has no namespace slash.""" if "/" in raw_name: return raw_name return f"{_DEFAULT_NAMESPACE}/{raw_name}" class ActorLoader: """Discover, validate, and cache actor YAML configurations. Parameters ---------- search_roots: Directories to scan for ``*.yaml`` / ``*.yml`` actor files. tool_registry: Optional tool registry used to verify tool references at load time. """ def __init__( self, search_roots: list[Path], tool_registry: ToolRegistry | None = None, ) -> None: self._search_roots = [Path(r) for r in search_roots] self._tool_registry = tool_registry self._actors: dict[str, _CacheEntry] = {} self._path_to_name: dict[Path, str] = {} self._lock = threading.RLock() self._warnings: list[str] = [] @property def warnings(self) -> list[str]: """Warnings emitted during the last discovery run.""" with self._lock: return list(self._warnings) def discover(self) -> list[ActorConfigSchema]: """Scan search roots and load or reload actor configs. Returns the list of all currently loaded actor configs. Raises ------ ValidationError When duplicate actor names are found across files or when a YAML file fails schema validation. """ with self._lock: self._warnings = [] found_files = self._collect_yaml_files() self._prune_deleted(found_files) pending: dict[str, list[tuple[Path, ActorConfigSchema]]] = {} errors: list[str] = [] for path in found_files: resolved = path.resolve() content = resolved.read_bytes() content_hash = _compute_hash(content) existing_name = self._path_to_name.get(resolved) if existing_name and existing_name in self._actors: entry = self._actors[existing_name] if entry.content_hash == content_hash: name = existing_name config = entry.config pending.setdefault(name, []).append((resolved, config)) continue try: raw = yaml.safe_load(content) except yaml.YAMLError as exc: mark = getattr(exc, "problem_mark", None) location = ( f" at line {mark.line + 1}, column {mark.column + 1}" if mark is not None else "" ) problem = getattr(exc, "problem", str(exc)) errors.append(f"Invalid YAML in {path.name}{location}: {problem}") continue if not isinstance(raw, dict): errors.append( f"Expected mapping in {path.name}, got {type(raw).__name__}" ) continue raw = self._apply_namespace_default(raw) try: config = ActorConfigSchema.model_validate(raw) except Exception as exc: from pydantic import ValidationError as PydanticValidationError if isinstance(exc, PydanticValidationError): field_errors = [] for err in exc.errors(): field_path = ".".join(str(loc) for loc in err["loc"]) field_errors.append(f" {field_path}: {err['msg']}") detail = "\n".join(field_errors) hint = ( " Hint: see docs/reference/actor_config.md " "for the correct schema format." ) errors.append( f"Schema validation failed for {path.name}:" f"\n{detail}\n{hint}" ) else: errors.append( f"Schema validation failed for {path.name}: {exc}" ) continue name = config.name pending.setdefault(name, []).append((resolved, config)) duplicate_msgs: list[str] = [] for name, entries in pending.items(): if len(entries) > 1: paths_str = ", ".join(str(p) for p, _ in entries) duplicate_msgs.append( f"Duplicate actor '{name}' found in: {paths_str}" ) if duplicate_msgs: all_errors = errors + duplicate_msgs raise ValidationError( "Actor discovery failed:\n" + "\n".join(all_errors), details={"duplicates": duplicate_msgs, "errors": errors}, ) if errors: raise ValidationError( "Actor discovery failed:\n" + "\n".join(errors), details={"errors": errors}, ) new_actors: dict[str, _CacheEntry] = {} new_path_map: dict[Path, str] = {} for name, entries in pending.items(): resolved_path, config = entries[0] content_hash = _compute_hash(resolved_path.read_bytes()) old_entry = self._actors.get(name) if old_entry and old_entry.content_hash == content_hash: new_actors[name] = old_entry else: entry = _CacheEntry( config=config, content_hash=content_hash, source_path=resolved_path, ) if old_entry: entry.load_count = old_entry.load_count + 1 new_actors[name] = entry new_path_map[resolved_path] = name self._resolve_tools(config) self._actors = new_actors self._path_to_name = new_path_map return [e.config for e in self._actors.values()] def get(self, name: str) -> ActorConfigSchema | None: """Retrieve a loaded actor by its namespaced name.""" with self._lock: entry = self._actors.get(name) return entry.config if entry else None def list_actors( self, namespace: str | None = None, ) -> list[ActorConfigSchema]: """List loaded actors with optional namespace filter. Parameters ---------- namespace: If provided, only return actors whose name starts with ``namespace/``. """ with self._lock: configs = [e.config for e in self._actors.values()] if namespace is not None: prefix = f"{namespace}/" configs = [c for c in configs if c.name.startswith(prefix)] return configs def clear(self) -> None: """Drop all cached actors and content hashes.""" with self._lock: self._actors.clear() self._path_to_name.clear() self._warnings.clear() def get_load_count(self, name: str) -> int: """Return how many times an actor was loaded from disk (for testing).""" with self._lock: entry = self._actors.get(name) return entry.load_count if entry else 0 def _collect_yaml_files(self) -> list[Path]: files: list[Path] = [] for root in self._search_roots: if not root.is_dir(): logger.warning("Search root is not a directory: %s", root) continue for path in root.rglob("*"): if path.is_file() and path.suffix in _YAML_SUFFIXES: files.append(path) return sorted(files) def _prune_deleted(self, current_files: list[Path]) -> None: resolved_set = {p.resolve() for p in current_files} stale_paths = [p for p in self._path_to_name if p not in resolved_set] for stale in stale_paths: name = self._path_to_name.pop(stale, None) if name: self._actors.pop(name, None) @staticmethod def _apply_namespace_default(raw: dict[str, Any]) -> dict[str, Any]: name = raw.get("name") if isinstance(name, str) and "/" not in name: raw["name"] = _normalize_name(name) return raw def _resolve_tools(self, config: ActorConfigSchema) -> None: if self._tool_registry is None: return for tool_ref in config.tools: if isinstance(tool_ref, str): spec = self._tool_registry.get(tool_ref) if spec is None: msg = ( f"Unresolved tool reference '{tool_ref}' " f"in actor '{config.name}'" ) logger.warning(msg) self._warnings.append(msg)