orchestrata_ultimate.kernel.config_manager

Reference › orchestrata_ultimate › kernel › config_manager

The configuration service of the kernel, SPEC-001 story US-009 (REQ-001-US-009-001 to -030). ConfigManager merges configuration from several sources into one typed, versioned view that can be validated, changed at runtime, diffed and rolled back. Nothing in the package imports it yet; it is covered by its own test files, and the curated kernel page lists it among the kernel services. The boot-time YAML loading used by the real deployment is in cli_main.

The data types: ConfigSource names where a value came from (default, file, environment, redis, runtime). ConfigValue wraps a value with its source, an encrypted flag and a last-updated time. ConfigSchema.validate checks a configuration and returns a flag with error messages. ConfigChange records each change with old and new value, and ConfigVersion is a snapshot of all values.

Loading: load_from_yaml reads a file, load_from_environment takes every variable with the prefix ORCHESTRATA_ and maps it to a key (the prefix is dropped, the rest is lower-cased and a double underscore becomes a dot), and load_from_redis reads keys under a prefix (orchestrata:config: by default) from a supplied Redis client. reload clears the values, reloads files, environment and Redis, validates, and restores a snapshot if validation or loading fails. start_file_watching polls file modification times at an interval and reloads changed YAML files.

Changing and reading: set validates through validate_before_apply and then _set_value, which skips unchanged values, updates or creates the ConfigValue, records a ConfigChange and calls the on_change callbacks, logging callback errors. get, get_typed and get_all read values. rollback restores a stored version, defaulting to the previous one. diff, get_change_history, dry_run, validate_compliance, export_yaml, export_json, generate_documentation and apply_template (which fills ${name} placeholders in a template file) support review. query_for_volk returns the keys under volk.<id>. merged with shared volk_type. keys. set_encrypted and get_decrypted store and read values flagged as encrypted.

At a glance

Class diagram

Dependency diagram

API

Classes

ConfigSource (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-006']) · Line 44

Configuration source types.

ConfigChange (dataclass)

Decorators: dataclass, traces(spec='001', story='US-009', reqs=['REQ-001-US-009-017']) · Line 56

Record of a configuration change (REQ-001-US-009-017).

ConfigVersion (dataclass)

Decorators: dataclass, traces(spec='001', story='US-009', reqs=['REQ-001-US-009-016']) · Line 86

Configuration version snapshot (REQ-001-US-009-016).

ConfigSchema (dataclass)

Decorators: dataclass, traces(spec='001', story='US-009', reqs=['REQ-001-US-009-004']) · Line 111

Schema definition for configuration validation (REQ-001-US-009-004).

ConfigValue

Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-006']) · Line 155

Typed configuration value with metadata.

ConfigManager

Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-001', 'REQ-001-US-009-004', 'REQ-001-US-009-006', 'REQ-001-US-… · Line 207

Main configuration management service (US-009).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/config_manager.py

# mypy: ignore-errors
"""Configuration Management for Orchestration Kernel.

Implements US-009: Configuration Management (REQ-001-US-009-001 to REQ-001-US-009-030).

Provides dynamic configuration management with:
- Multi-source loading (YAML, env, Redis)
- Live reload without restart
- Versioning and change tracking
- Validation and rollback
"""

import asyncio
import hashlib
import inspect
import json
import logging
import os
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import Any, TypeVar

import yaml

from orchestrata_ultimate.core.base.traces import traces

# Module-level logger
_logger = logging.getLogger(__name__)

T = TypeVar("T")


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-017"])
def _utc_now() -> datetime:
    """Get current UTC time with timezone info."""
    return datetime.now(UTC)


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
class ConfigSource(Enum):
    """Configuration source types."""

    DEFAULT = "default"
    FILE = "file"
    ENVIRONMENT = "environment"
    REDIS = "redis"
    RUNTIME = "runtime"


@dataclass
@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-017"])
class ConfigChange:
    """Record of a configuration change (REQ-001-US-009-017)."""

    change_id: str
    timestamp: datetime
    source: ConfigSource
    key: str
    old_value: Any
    new_value: Any
    applied: bool
    rolled_back: bool = False
    reason: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "change_id": self.change_id,
            "timestamp": self.timestamp.isoformat(),
            "source": self.source.value,
            "key": self.key,
            "old_value": self.old_value,
            "new_value": self.new_value,
            "applied": self.applied,
            "rolled_back": self.rolled_back,
            "reason": self.reason,
        }


@dataclass
@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-016"])
class ConfigVersion:
    """Configuration version snapshot (REQ-001-US-009-016)."""

    version: int
    timestamp: datetime
    config_hash: str
    config_data: dict[str, Any]
    source: ConfigSource
    comment: str | None = None

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-016"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "version": self.version,
            "timestamp": self.timestamp.isoformat(),
            "config_hash": self.config_hash,
            "config_data": self.config_data,
            "source": self.source.value,
            "comment": self.comment,
        }


@dataclass
@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
class ConfigSchema:
    """Schema definition for configuration validation (REQ-001-US-009-004)."""

    required_keys: list[str] = field(default_factory=list)
    optional_keys: list[str] = field(default_factory=list)
    key_types: dict[str, type] = field(default_factory=dict)
    key_validators: dict[str, Callable[[Any], bool]] = field(default_factory=dict)
    key_defaults: dict[str, Any] = field(default_factory=dict)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def validate(self, config: dict[str, Any]) -> tuple[bool, list[str]]:
        """Validate configuration against schema.

        Returns:
            Tuple of (is_valid, list of error messages)
        """
        errors = []

        # Check required keys
        for key in self.required_keys:
            if key not in config:
                errors.append(f"Missing required key: {key}")

        # Check types
        for key, expected_type in self.key_types.items():
            if key in config and not isinstance(config[key], expected_type):
                errors.append(
                    f"Invalid type for {key}: expected {expected_type.__name__}, "
                    f"got {type(config[key]).__name__}"
                )

        # Run custom validators
        for key, validator in self.key_validators.items():
            if key in config:
                try:
                    if not validator(config[key]):
                        errors.append(f"Validation failed for key: {key}")
                except Exception as e:
                    errors.append(f"Validator error for {key}: {e}")

        return len(errors) == 0, errors


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
class ConfigValue[T]:
    """Typed configuration value with metadata."""

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-009"])
    def __init__(
        self,
        value: T,
        source: ConfigSource = ConfigSource.DEFAULT,
        encrypted: bool = False,
    ) -> None:
        self._value = value
        self._source = source
        self._encrypted = encrypted
        self._last_updated = _utc_now()

    @property
    def value(self) -> T:
        return self._value

    @property
    def source(self) -> ConfigSource:
        return self._source

    @property
    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-009"])
    def encrypted(self) -> bool:
        return self._encrypted

    @property
    def last_updated(self) -> datetime:
        return self._last_updated

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-012"])
    def update(self, value: T, source: ConfigSource) -> None:
        """Update the value."""
        self._value = value
        self._source = source
        self._last_updated = _utc_now()


@traces(
    spec="001",
    story="US-009",
    reqs=[
        "REQ-001-US-009-001",
        "REQ-001-US-009-004",
        "REQ-001-US-009-006",
        "REQ-001-US-009-011",
        "REQ-001-US-009-016",
        "REQ-001-US-009-017",
    ],
)
class ConfigManager:
    """Main configuration management service (US-009).

    Provides comprehensive configuration management with:
    - Multi-source loading (REQ-001-US-009-001 to REQ-001-US-009-003)
    - Schema validation (REQ-001-US-009-004)
    - Defaults and overrides (REQ-001-US-009-005, REQ-001-US-009-006)
    - Dynamic reload (REQ-001-US-009-011)
    - Versioning (REQ-001-US-009-016)
    - Change tracking (REQ-001-US-009-017)
    """

    def __init__(
        self,
        config_dir: Path | None = None,
        schema: ConfigSchema | None = None,
        redis_client: Any | None = None,
        redis_key_prefix: str = "orchestrata:config:",
    ) -> None:
        self.config_dir = config_dir or Path("configs")
        self.schema = schema or ConfigSchema()
        self._redis = redis_client
        self._redis_prefix = redis_key_prefix

        # Configuration state
        self._config: dict[str, ConfigValue] = {}
        self._defaults: dict[str, Any] = {}
        self._versions: list[ConfigVersion] = []
        self._changes: list[ConfigChange] = []
        self._current_version = 0

        # Callbacks for change notifications (REQ-001-US-009-015)
        self._change_callbacks: list[Callable[[str, Any, Any], None]] = []

        # Lock for thread safety
        self._lock = asyncio.Lock()

        # File watching
        self._watch_task: asyncio.Task | None = None
        self._file_mtimes: dict[Path, float] = {}

    # ===== Configuration Loading (REQ-001-US-009-001 to REQ-001-US-009-003) =====

    @traces(
        spec="001",
        story="US-009",
        reqs=["REQ-001-US-009-001", "REQ-001-US-009-007"],
    )
    async def load_from_yaml(self, file_path: Path) -> bool:
        """Load configuration from YAML file (REQ-001-US-009-001)."""
        try:
            if not file_path.exists():
                _logger.warning(f"Config file not found: {file_path}")
                return False

            with open(file_path, encoding="utf-8") as f:
                data = yaml.safe_load(f) or {}

            async with self._lock:
                for key, value in data.items():
                    await self._set_value(key, value, ConfigSource.FILE)

            _logger.info(f"Loaded configuration from {file_path}")
            return True

        except yaml.YAMLError as e:
            _logger.error(f"YAML parse error in {file_path}: {e}")
            return False
        except Exception as e:
            _logger.error(f"Error loading config from {file_path}: {e}")
            return False

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    async def load_from_environment(
        self, prefix: str = "ORCHESTRATA_", environ: dict[str, str] | None = None
    ) -> int:
        """Load configuration from environment variables (REQ-001-US-009-002).

        Args:
            prefix: Prefix for environment variables to load.
            environ: Optional environment mapping (defaults to os.environ).

        Returns:
            Number of variables loaded.
        """
        count = 0
        env = environ or os.environ

        async with self._lock:
            for key, value in env.items():
                if key.startswith(prefix):
                    config_key = key[len(prefix) :].lower().replace("__", ".")
                    parsed_value = self._parse_env_value(value)
                    await self._set_value(
                        config_key, parsed_value, ConfigSource.ENVIRONMENT
                    )
                    count += 1

        _logger.info(f"Loaded {count} configuration values from environment")
        return count

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-003"])
    async def load_from_redis(self) -> bool:
        """Load configuration from Redis (REQ-001-US-009-003)."""
        if self._redis is None:
            _logger.warning("Redis client not configured")
            return False

        try:
            # Get all config keys
            pattern = f"{self._redis_prefix}*"
            keys = await self._redis_keys(pattern)

            async with self._lock:
                for key in keys:
                    config_key = key.decode() if isinstance(key, bytes) else key
                    config_key = config_key[len(self._redis_prefix) :]
                    value = await self._redis_get(key)
                    if value is not None:
                        parsed = self._parse_redis_value(value)
                        await self._set_value(config_key, parsed, ConfigSource.REDIS)

            _logger.info(f"Loaded {len(keys)} configuration values from Redis")
            return True

        except Exception as e:
            _logger.error(f"Error loading config from Redis: {e}")
            return False

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-003"])
    async def _redis_keys(self, pattern: str) -> list[bytes | str]:
        """Get Redis keys matching pattern."""
        # self._redis is duck-typed (Any | None) since it accepts either a
        # sync or async redis-like client; the None-check is required (a
        # bare hasattr() doesn't narrow away the None branch for pyright).
        if self._redis is None or not hasattr(self._redis, "keys"):
            return []
        if inspect.iscoroutinefunction(self._redis.keys):
            return await self._redis.keys(pattern)
        return self._redis.keys(pattern)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-003"])
    async def _redis_get(self, key: bytes | str) -> bytes | None:
        """Get value from Redis."""
        if self._redis is None or not hasattr(self._redis, "get"):
            return None
        if inspect.iscoroutinefunction(self._redis.get):
            return await self._redis.get(key)
        return self._redis.get(key)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-010"])
    async def _redis_set(self, key: str, value: str) -> bool:
        """Set value in Redis."""
        if self._redis is None or not hasattr(self._redis, "set"):
            return False
        if inspect.iscoroutinefunction(self._redis.set):
            await self._redis.set(key, value)
        else:
            self._redis.set(key, value)
        return True

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    def _parse_env_value(self, value: str) -> Any:
        """Parse environment variable value to appropriate type."""
        # Boolean
        if value.lower() in ("true", "yes", "1"):
            return True
        if value.lower() in ("false", "no", "0"):
            return False

        # Integer
        try:
            return int(value)
        except ValueError as exc:
            _logger.debug("int parse: %s", exc)

        # Float
        try:
            return float(value)
        except ValueError as exc:
            _logger.debug("float parse: %s", exc)

        # JSON
        try:
            return json.loads(value)
        except json.JSONDecodeError as exc:
            _logger.debug("JSON parse: %s", exc)

        # String
        return value

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-003"])
    def _parse_redis_value(self, value: bytes) -> Any:
        """Parse Redis value to appropriate type."""
        str_value = value.decode() if isinstance(value, bytes) else value
        return self._parse_env_value(str_value)

    # ===== Schema Validation (REQ-001-US-009-004) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    async def validate(self) -> tuple[bool, list[str]]:
        """Validate current configuration against schema."""
        config_dict = await self.get_all()
        return self.schema.validate(config_dict)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-013"])
    async def validate_before_apply(
        self,
        key: str,
        value: Any,
    ) -> tuple[bool, list[str]]:
        """Validate a single value before applying (REQ-001-US-009-013)."""
        errors = []

        # Check type if defined
        if key in self.schema.key_types:
            expected_type = self.schema.key_types[key]
            if not isinstance(value, expected_type):
                errors.append(
                    f"Invalid type for {key}: expected {expected_type.__name__}"
                )

        # Run custom validator if defined
        if key in self.schema.key_validators:
            try:
                if not self.schema.key_validators[key](value):
                    errors.append(f"Validation failed for key: {key}")
            except Exception as e:
                errors.append(f"Validator error for {key}: {e}")

        return len(errors) == 0, errors

    # ===== Defaults and Overrides (REQ-001-US-009-005, REQ-001-US-009-006) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-005"])
    def set_defaults(self, defaults: dict[str, Any]) -> None:
        """Set default configuration values (REQ-001-US-009-005)."""
        self._defaults.update(defaults)
        self.schema.key_defaults.update(defaults)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-024"])
    async def get(
        self,
        key: str,
        default: T | None = None,
    ) -> T | None:
        """Get configuration value with fallback to default."""
        async with self._lock:
            if key in self._config:
                return self._config[key].value

            # Check schema defaults
            if key in self._defaults:
                return self._defaults[key]

            return default

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-024"])
    async def get_typed(
        self,
        key: str,
        value_type: type[T],
        default: T | None = None,
    ) -> T | None:
        """Get configuration value with type checking."""
        value = await self.get(key, default)
        if value is not None and not isinstance(value, value_type):
            _logger.warning(
                f"Config value {key} has wrong type: expected {value_type.__name__}"
            )
            return default
        return value

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-021"])
    async def get_all(self) -> dict[str, Any]:
        """Get all configuration values."""
        async with self._lock:
            result = dict(self._defaults)
            for key, config_value in self._config.items():
                result[key] = config_value.value
            return result

    # ===== Dynamic Configuration (REQ-001-US-009-011 to REQ-001-US-009-015) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-012"])
    async def set(
        self,
        key: str,
        value: Any,
        source: ConfigSource = ConfigSource.RUNTIME,
        validate: bool = True,
    ) -> bool:
        """Set configuration value dynamically (REQ-001-US-009-012).

        Returns:
            True if value was set successfully.
        """
        # Validate before applying (REQ-001-US-009-013)
        if validate:
            is_valid, errors = await self.validate_before_apply(key, value)
            if not is_valid:
                _logger.error(f"Validation failed for {key}: {errors}")
                return False

        async with self._lock:
            return await self._set_value(key, value, source)

    @traces(
        spec="001",
        story="US-009",
        reqs=["REQ-001-US-009-012", "REQ-001-US-009-017", "REQ-001-US-009-018"],
    )
    async def _set_value(
        self,
        key: str,
        value: Any,
        source: ConfigSource,
    ) -> bool:
        """Internal method to set configuration value."""
        old_value = None
        if key in self._config:
            old_value = self._config[key].value

        # Skip if value unchanged
        if old_value == value:
            return True

        # Update value
        if key in self._config:
            self._config[key].update(value, source)
        else:
            self._config[key] = ConfigValue(value, source)

        # Record change (REQ-001-US-009-017, REQ-001-US-009-018)
        change = ConfigChange(
            change_id=f"chg-{len(self._changes):05d}",
            timestamp=_utc_now(),
            source=source,
            key=key,
            old_value=old_value,
            new_value=value,
            applied=True,
        )
        self._changes.append(change)
        _logger.info(f"Config change: {key} = {value} (from {source.value})")

        # Notify callbacks (REQ-001-US-009-015)
        for callback in self._change_callbacks:
            try:
                callback(key, old_value, value)
            except Exception as e:
                _logger.error(f"Error in config change callback: {e}")

        return True

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def reload(self, environ: dict[str, str] | None = None) -> bool:
        """Reload configuration from all sources (REQ-001-US-009-011).

        Args:
            environ: Optional environment mapping for testing.
        """
        _logger.info("Reloading configuration...")

        # Create snapshot for potential rollback
        snapshot = await self._create_snapshot()

        try:
            # Clear current config (keep defaults)
            async with self._lock:
                self._config.clear()

            # Reload from sources in priority order (REQ-001-US-009-006)
            # 1. Defaults (already set)
            # 2. YAML files
            for yaml_file in self.config_dir.glob("*.yaml"):
                await self.load_from_yaml(yaml_file)
            for yaml_file in self.config_dir.glob("*.yml"):
                await self.load_from_yaml(yaml_file)

            # 3. Environment variables
            await self.load_from_environment(environ=environ)

            # 4. Redis (highest priority)
            if self._redis:
                await self.load_from_redis()

            # Validate (REQ-001-US-009-013)
            is_valid, errors = await self.validate()
            if not is_valid:
                _logger.error(f"Configuration validation failed: {errors}")
                await self._restore_snapshot(snapshot)
                return False

            # Save version
            await self._save_version("Reloaded configuration")

            _logger.info("Configuration reloaded successfully")
            return True

        except Exception as e:
            _logger.error(f"Error reloading configuration: {e}")
            await self._restore_snapshot(snapshot)
            return False

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-014"])
    async def rollback(self, version: int | None = None) -> bool:
        """Rollback to a previous version (REQ-001-US-009-014)."""
        if not self._versions:
            _logger.warning("No versions available for rollback")
            return False

        target_version = version if version is not None else self._current_version - 1
        target = None

        for v in self._versions:
            if v.version == target_version:
                target = v
                break

        if target is None:
            _logger.error(f"Version {target_version} not found")
            return False

        async with self._lock:
            self._config.clear()
            for key, value in target.config_data.items():
                self._config[key] = ConfigValue(value, ConfigSource.RUNTIME)

        _logger.info(f"Rolled back to version {target_version}")
        return True

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-015"])
    def on_change(self, callback: Callable[[str, Any, Any], None]) -> None:
        """Register callback for configuration changes (REQ-001-US-009-015)."""
        self._change_callbacks.append(callback)

    # ===== Versioning (REQ-001-US-009-016, REQ-001-US-009-017) =====

    @traces(
        spec="001", story="US-009", reqs=["REQ-001-US-009-014", "REQ-001-US-009-016"]
    )
    async def _create_snapshot(self) -> dict[str, Any]:
        """Create snapshot of current configuration."""
        async with self._lock:
            return {key: cv.value for key, cv in self._config.items()}

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-014"])
    async def _restore_snapshot(self, snapshot: dict[str, Any]) -> None:
        """Restore configuration from snapshot."""
        async with self._lock:
            self._config.clear()
            for key, value in snapshot.items():
                self._config[key] = ConfigValue(value, ConfigSource.RUNTIME)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-016"])
    async def _save_version(self, comment: str | None = None) -> None:
        """Save current configuration as a new version."""
        config_data = await self._create_snapshot()
        config_hash = hashlib.sha256(
            json.dumps(config_data, sort_keys=True).encode()
        ).hexdigest()[:12]

        self._current_version += 1
        version = ConfigVersion(
            version=self._current_version,
            timestamp=_utc_now(),
            config_hash=config_hash,
            config_data=config_data,
            source=ConfigSource.RUNTIME,
            comment=comment,
        )
        self._versions.append(version)

        # Keep only last 100 versions
        if len(self._versions) > 100:
            self._versions = self._versions[-100:]

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-016"])
    async def get_version(self, version: int | None = None) -> ConfigVersion | None:
        """Get a specific version or current version."""
        if version is None:
            version = self._current_version

        for v in self._versions:
            if v.version == version:
                return v
        return None

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-017"])
    async def get_change_history(
        self,
        key: str | None = None,
        limit: int = 100,
    ) -> list[ConfigChange]:
        """Get configuration change history (REQ-001-US-009-017)."""
        changes = self._changes
        if key:
            changes = [c for c in changes if c.key == key]
        return changes[-limit:]

    # ===== Query Methods (REQ-001-US-009-021 to REQ-001-US-009-030) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-021"])
    async def query_current(self) -> dict[str, Any]:
        """Query current configuration (REQ-001-US-009-021)."""
        return await self.get_all()

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-022"])
    async def query_history(
        self,
        limit: int = 100,
    ) -> list[dict[str, Any]]:
        """Query configuration history (REQ-001-US-009-022)."""
        return [v.to_dict() for v in self._versions[-limit:]]

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-023"])
    async def query_for_volk(self, volk_id: str) -> dict[str, Any]:
        """Query configuration for specific Volk (REQ-001-US-009-023)."""
        all_config = await self.get_all()
        volk_config = {}

        # Get volk-specific config
        volk_prefix = f"volk.{volk_id}."
        for key, value in all_config.items():
            if key.startswith(volk_prefix):
                volk_config[key[len(volk_prefix) :]] = value

        # Merge with global config for volk type
        volk_type_prefix = "volk_type."
        for key, value in all_config.items():
            if key.startswith(volk_type_prefix):
                short_key = key[len(volk_type_prefix) :]
                if short_key not in volk_config:
                    volk_config[short_key] = value

        return volk_config

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-024"])
    async def query_by_key(self, key_pattern: str) -> dict[str, Any]:
        """Query configuration by key pattern (REQ-001-US-009-024)."""
        all_config = await self.get_all()
        pattern = re.compile(key_pattern.replace("*", ".*"))
        return {k: v for k, v in all_config.items() if pattern.match(k)}

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-025"])
    async def export_yaml(self, output_path: Path | None = None) -> str:
        """Export configuration as YAML (REQ-001-US-009-025)."""
        config = await self.get_all()
        yaml_str = yaml.dump(config, default_flow_style=False, sort_keys=True)

        if output_path:
            output_path.write_text(yaml_str)

        return yaml_str

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-026"])
    async def export_json(self, output_path: Path | None = None) -> str:
        """Export configuration as JSON (REQ-001-US-009-026)."""
        config = await self.get_all()
        json_str = json.dumps(config, indent=2, sort_keys=True)

        if output_path:
            output_path.write_text(json_str)

        return json_str

    @traces(
        spec="001",
        story="US-009",
        reqs=["REQ-001-US-009-020", "REQ-001-US-009-027"],
    )
    async def diff(
        self,
        version1: int,
        version2: int | None = None,
    ) -> dict[str, dict[str, Any] | str]:
        """Compare two configuration versions (REQ-001-US-009-020, REQ-001-US-009-027)."""
        from typing import cast

        v1 = await self.get_version(version1)
        v2 = await self.get_version(version2)

        if v1 is None or v2 is None:
            return {"error": "Version not found"}

        diff: dict[str, dict[str, Any]] = {
            "added": {},
            "removed": {},
            "changed": {},
        }

        all_keys = set(v1.config_data.keys()) | set(v2.config_data.keys())

        for key in all_keys:
            in_v1 = key in v1.config_data
            in_v2 = key in v2.config_data

            if in_v1 and not in_v2:
                diff["removed"][key] = v1.config_data[key]
            elif not in_v1 and in_v2:
                diff["added"][key] = v2.config_data[key]
            elif v1.config_data[key] != v2.config_data[key]:
                diff["changed"][key] = {
                    "from": v1.config_data[key],
                    "to": v2.config_data[key],
                }

        # dict's value type is invariant, so the narrower
        # dict[str, dict[str, Any]] built above needs an explicit cast to
        # the wider return type shared with the {"error": ...} early return.
        return cast(dict[str, dict[str, Any] | str], diff)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-028"])
    async def validate_compliance(
        self,
        rules: dict[str, Callable[[Any], bool]],
    ) -> dict[str, bool]:
        """Validate configuration compliance (REQ-001-US-009-028)."""
        config = await self.get_all()
        results = {}

        for rule_name, rule_fn in rules.items():
            try:
                results[rule_name] = rule_fn(config)
            except Exception as e:
                _logger.error(f"Compliance rule {rule_name} failed: {e}")
                results[rule_name] = False

        return results

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-029"])
    async def generate_documentation(self) -> str:
        """Generate configuration documentation (REQ-001-US-009-029)."""
        config = await self.get_all()

        doc_lines = [
            "# Configuration Documentation",
            "",
            f"Generated: {_utc_now().isoformat()}",
            f"Version: {self._current_version}",
            "",
            "## Current Configuration",
            "",
        ]

        # Group by prefix
        groups: dict[str, dict[str, Any]] = {}
        for key, value in sorted(config.items()):
            parts = key.split(".")
            group = parts[0] if len(parts) > 1 else "general"
            if group not in groups:
                groups[group] = {}
            groups[group][key] = value

        for group_name, group_config in sorted(groups.items()):
            doc_lines.append(f"### {group_name.title()}")
            doc_lines.append("")
            doc_lines.append("| Key | Value | Type |")
            doc_lines.append("|-----|-------|------|")

            for key, value in sorted(group_config.items()):
                value_str = str(value)
                if len(value_str) > 50:
                    value_str = value_str[:47] + "..."
                doc_lines.append(f"| {key} | {value_str} | {type(value).__name__} |")

            doc_lines.append("")

        # Add schema info
        if self.schema.required_keys:
            doc_lines.append("## Required Keys")
            doc_lines.append("")
            for key in self.schema.required_keys:
                doc_lines.append(f"- `{key}`")
            doc_lines.append("")

        return "\n".join(doc_lines)

    # ===== Templates (REQ-001-US-009-030) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-030"])
    async def apply_template(
        self,
        template_name: str,
        template_values: dict[str, Any] | None = None,
    ) -> bool:
        """Apply a configuration template (REQ-001-US-009-030)."""
        template_path = self.config_dir / "templates" / f"{template_name}.yaml"

        if not template_path.exists():
            _logger.error(f"Template not found: {template_name}")
            return False

        try:
            with open(template_path, encoding="utf-8") as f:
                template_content = f.read()

            # Substitute template variables
            if template_values:
                for key, value in template_values.items():
                    template_content = template_content.replace(
                        f"${{{key}}}", str(value)
                    )

            template_config = yaml.safe_load(template_content) or {}

            async with self._lock:
                for key, value in template_config.items():
                    await self._set_value(key, value, ConfigSource.FILE)

            await self._save_version(f"Applied template: {template_name}")
            _logger.info(f"Applied configuration template: {template_name}")
            return True

        except Exception as e:
            _logger.error(f"Error applying template {template_name}: {e}")
            return False

    # ===== Persistence (REQ-001-US-009-010) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-010"])
    async def persist_to_redis(self) -> bool:
        """Persist current configuration to Redis (REQ-001-US-009-010)."""
        if self._redis is None:
            return False

        try:
            config = await self.get_all()
            for key, value in config.items():
                redis_key = f"{self._redis_prefix}{key}"
                await self._redis_set(redis_key, json.dumps(value))

            _logger.info("Configuration persisted to Redis")
            return True

        except Exception as e:
            _logger.error(f"Error persisting to Redis: {e}")
            return False

    # ===== File Watching =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def start_file_watching(self, interval: float = 5.0) -> None:
        """Start watching configuration files for changes."""
        if self._watch_task is not None:
            return

        self._watch_task = asyncio.create_task(self._watch_files(interval))
        _logger.info("Started configuration file watching")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def stop_file_watching(self) -> None:
        """Stop watching configuration files."""
        if self._watch_task:
            self._watch_task.cancel()
            try:
                await self._watch_task
            except asyncio.CancelledError:
                _logger.debug("config watch task cancelled")
            self._watch_task = None
        _logger.info("Stopped configuration file watching")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def _watch_files(self, interval: float) -> None:
        """Watch configuration files for changes."""
        while True:
            try:
                await asyncio.sleep(interval)

                for yaml_file in self.config_dir.glob("*.yaml"):
                    mtime = yaml_file.stat().st_mtime
                    if yaml_file in self._file_mtimes:
                        if mtime > self._file_mtimes[yaml_file]:
                            _logger.info(f"Detected change in {yaml_file}")
                            await self.load_from_yaml(yaml_file)
                    self._file_mtimes[yaml_file] = mtime

            except asyncio.CancelledError:
                break
            except Exception as e:
                _logger.error(f"Error watching files: {e}")

    # ===== Encrypted Values (REQ-001-US-009-009) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-009"])
    async def set_encrypted(
        self,
        key: str,
        value: str,
        encryption_key: str | None = None,
    ) -> bool:
        """Set an encrypted configuration value (REQ-001-US-009-009)."""
        # Simple base64 encoding for demonstration
        # In production, use proper encryption (Fernet, etc.)
        import base64

        try:
            encoded = base64.b64encode(value.encode()).decode()
            async with self._lock:
                self._config[key] = ConfigValue(
                    f"encrypted:{encoded}",
                    ConfigSource.RUNTIME,
                    encrypted=True,
                )
            return True
        except Exception as e:
            _logger.error(f"Error encrypting value: {e}")
            return False

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-009"])
    async def get_decrypted(
        self,
        key: str,
        encryption_key: str | None = None,
    ) -> str | None:
        """Get a decrypted configuration value."""
        import base64

        async with self._lock:
            if key not in self._config:
                return None

            value = self._config[key].value
            if not isinstance(value, str) or not value.startswith("encrypted:"):
                return value

            try:
                encoded = value[len("encrypted:") :]
                return base64.b64decode(encoded).decode()
            except Exception as e:
                _logger.error(f"Error decrypting value: {e}")
                return None

    # ===== Dry Run Mode (REQ-001-US-009-019) =====

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-019"])
    async def dry_run(
        self,
        changes: dict[str, Any],
    ) -> dict[str, Any]:
        """Test configuration changes without applying (REQ-001-US-009-019)."""
        # Create a copy of current config
        current = await self._create_snapshot()
        test_config = dict(current)
        test_config.update(changes)

        # Validate
        errors = []
        for key, value in changes.items():
            is_valid, key_errors = await self.validate_before_apply(key, value)
            if not is_valid:
                errors.extend(key_errors)

        # Check full config validation
        full_valid, full_errors = self.schema.validate(test_config)
        if not full_valid:
            errors.extend(full_errors)

        # Compute diff
        diff = {
            "added": {},
            "modified": {},
        }
        for key, value in changes.items():
            if key not in current:
                diff["added"][key] = value
            elif current[key] != value:
                diff["modified"][key] = {"from": current[key], "to": value}

        return {
            "valid": len(errors) == 0,
            "errors": errors,
            "diff": diff,
            "resulting_config": test_config,
        }