config_helpers.py

"""Configuration resolution utilities for hierarchical settings."""

import logging
from typing import Any, cast

from src import models
from src.config import settings
from src.schemas import (
    MessageConfiguration,
    ResolvedConfiguration,
)

logger = logging.getLogger(__name__)

_NONE_OVERRIDE_PATHS: set[tuple[str, ...]] = {
    ("summary", "custom_instructions"),
}


def deep_update(
    base: dict[str, Any],
    update: dict[str, Any],
    path: tuple[str, ...] = (),
) -> None:
    """
    Recursive update of a dictionary.
    Skips None values unless None explicitly clears a nullable field.
    """
    for key, value in update.items():
        current_path = (*path, key)

        if value is None:
            if current_path in _NONE_OVERRIDE_PATHS:
                base[key] = None
            continue

        if isinstance(value, dict) and key in base and isinstance(base[key], dict):
            deep_update(
                cast(dict[str, Any], base[key]),
                cast(dict[str, Any], value),
                current_path,
            )
        else:
            base[key] = value


def normalize_configuration_dict(raw: dict[str, Any]) -> dict[str, Any]:
    """
    Normalize a workspace/session/message configuration dict to match the current
    `ResolvedConfiguration` schema.

    This function exists to preserve backwards compatibility with older clients/tests
    that used legacy configuration keys (e.g. `deriver.enabled` or `skip_deriver`).

    Behavior:
    - If `reasoning.enabled` is not explicitly set, but `deriver.enabled` is present,
      `reasoning.enabled` is derived from `deriver.enabled`.
    - If `skip_deriver` is explicitly `True`, `reasoning.enabled` is forced to `False`
      unless `reasoning.enabled` was already explicitly set.
    - Legacy keys are removed from the returned dict to avoid polluting the resolved
      configuration with unused fields.
    """
    normalized: dict[str, Any] = dict(raw)

    reasoning_raw = normalized.get("reasoning")
    reasoning_present = "reasoning" in normalized
    reasoning: dict[str, Any]
    if isinstance(reasoning_raw, dict):
        reasoning = dict(cast(dict[str, Any], reasoning_raw))
    else:
        reasoning = {}
    reasoning_enabled_explicit = reasoning.get("enabled") is not None

    if not reasoning_enabled_explicit:
        deriver_raw = normalized.get("deriver")
        deriver: dict[str, Any]
        if isinstance(deriver_raw, dict):
            deriver = dict(cast(dict[str, Any], deriver_raw))
        else:
            deriver = {}
        if deriver.get("enabled") is not None:
            reasoning["enabled"] = bool(deriver["enabled"])

    if not reasoning_enabled_explicit and normalized.get("skip_deriver") is True:
        reasoning["enabled"] = False

    if reasoning_present or reasoning:
        normalized["reasoning"] = reasoning

    normalized.pop("deriver", None)
    normalized.pop("skip_deriver", None)

    return normalized


def get_configuration(
    message_configuration: MessageConfiguration | None,
    session: models.Session | None,
    workspace: models.Workspace | None = None,
) -> ResolvedConfiguration:
    """
    Resolve session configuration with hierarchical fallback.

    Resolution hierarchy:
    1. Message configuration
    2. Session configuration
    3. Workspace configuration
    4. Global defaults from settings

    Args:
        message_configuration: Optional message configuration
        session: Optional session model
        workspace: Optional workspace model

    Returns:
        ResolvedConfiguration
    """
    # Start with defaults
    config_dict: dict[str, Any] = {
        "reasoning": {
            "enabled": settings.DERIVER.ENABLED,
            "custom_instructions": None,
        },
        "peer_card": {
            "use": settings.PEER_CARD.ENABLED,
            "create": settings.PEER_CARD.ENABLED,
        },
        "summary": {
            "enabled": settings.SUMMARY.ENABLED,
            "messages_per_short_summary": settings.SUMMARY.MESSAGES_PER_SHORT_SUMMARY,
            "messages_per_long_summary": settings.SUMMARY.MESSAGES_PER_LONG_SUMMARY,
            "custom_instructions": None,
        },
        "dream": {"enabled": settings.DREAM.ENABLED},
    }

    # Apply overrides in order (Workspace -> Session -> Message)
    # Note: deep_update modifies config_dict in place

    if workspace is not None:
        deep_update(config_dict, normalize_configuration_dict(workspace.configuration))

    if session is not None:
        deep_update(config_dict, normalize_configuration_dict(session.configuration))

    if message_configuration is not None:
        deep_update(
            config_dict,
            normalize_configuration_dict(
                message_configuration.model_dump(exclude_unset=True)
            ),
        )

    return ResolvedConfiguration(**config_dict)