"""Root configuration model and helpers."""

from __future__ import annotations

import re
from collections import deque
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast

import yaml
from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    PrivateAttr,
    ValidationError,
    ValidationInfo,
    field_validator,
    model_validator,
)

from mindroom import yaml_io
from mindroom.agent_policy import (
    build_agent_policy_seeds,
    get_agent_delegation_closure,
    get_unsupported_team_agents,
    resolve_agent_policy_from_data,
    resolve_private_knowledge_base_agent,
    unsupported_team_agent_message,
)
from mindroom.config.agent import AgentConfig, CultureConfig, RoomConfig, TeamConfig  # noqa: TC001
from mindroom.config.approval import ToolApprovalConfig
from mindroom.config.auth import AuthorizationConfig
from mindroom.config.calls import CallsConfig, CascadedCallProfile
from mindroom.config.entity_view import ResolvedEntityView
from mindroom.config.external_trigger_policy import ExternalTriggerPolicyConfig
from mindroom.config.knowledge import KnowledgeBaseConfig
from mindroom.config.matrix import (
    CacheConfig,
    MatrixRoomAccessConfig,
    MatrixSpaceConfig,
    MatrixSyncConfig,
    MindRoomUserConfig,
)
from mindroom.config.memory import MemoryBackend, MemoryConfig, MemorySearchConfig
from mindroom.config.models import (
    CompactionConfig,
    CompactionOverrideConfig,
    DebugConfig,
    DefaultsConfig,
    EffectiveToolConfig,
    ModelConfig,
    RouterConfig,
    ToolConfigEntry,
)
from mindroom.config.plugin import PluginEntryConfig  # noqa: TC001
from mindroom.config.runtime_overlays import (
    apply_runtime_approved_egress_overlay,
    strip_runtime_approved_egress_overlay_from_dump,
)
from mindroom.config.tool_entries import raw_tool_entry_name_and_lazy_flag_fields, raw_tools_entries
from mindroom.config.voice import VoiceConfig
from mindroom.config.yaml_includes import ConfigIncludeError, attach_partial_source_files, load_yaml_config_source
from mindroom.constants import (
    DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    ROUTER_AGENT_NAME,
    RuntimePaths,
    config_relative_path,
    matrix_state_file,
    resolve_config_relative_path,
    runtime_matrix_homeserver,
)
from mindroom.git_urls import credential_free_repo_url

# config layer loads BEFORE the history runtime; import leaf types so config load does not drag in agents+tools.
from mindroom.history.types import HistoryPolicy, ResolvedHistorySettings
from mindroom.logging_config import get_logger
from mindroom.matrix_identifiers import (
    extract_server_name_from_homeserver,
    managed_room_alias_localpart,
    managed_space_alias_localpart,
)
from mindroom.mcp.config import MCPServerConfig, normalize_mcp_server_id
from mindroom.prompt_templates import render_prompt_template, validate_prompt_template_fields
from mindroom.prompts import PROMPT_DEFAULT_NAMES, PROMPT_DEFAULTS
from mindroom.room_thread_modes import resolve_room_thread_mode_override
from mindroom.runtime_env_policy import SANDBOX_RUNTIME_ENV_BY_KEY
from mindroom.thread_models import resolve_thread_model_override
from mindroom.tool_system.plugin_imports import PluginValidationError
from mindroom.tool_system.worker_routing import unsupported_shared_only_integration_names
from mindroom.workspaces import validate_workspace_template_dir

if TYPE_CHECKING:
    from collections.abc import Collection

    from mindroom.tool_system.catalog import ToolValidationInfo
    from mindroom.tool_system.worker_routing import WorkerScope

# Keep synchronized with todo_poke._SAFE_ASSIGNEE_PATTERN without importing runtime tools into config.
_AGENT_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9_]+$")
_RESERVED_ENTITY_NAMES = frozenset({ROUTER_AGENT_NAME, "user"})
_DEFER_PROHIBITED_CONTROL_TOOLS = frozenset({"delegate", "dynamic_tools", "external_trigger_manager", "self_config"})
_OPENCLAW_COMPAT_PRESET_TOOLS: tuple[str, ...] = (
    "shell",
    "coding",
    "duckduckgo",
    "website",
    "browser",
    "scheduler",
    "subagents",
    "matrix_message",
)


logger = get_logger(__name__)


def _persisted_entity_account_usernames(runtime_paths: RuntimePaths) -> dict[str, str]:
    state_file = matrix_state_file(runtime_paths=runtime_paths)
    if not state_file.exists():
        return {}
    data = yaml_io.safe_load(state_file.read_text(encoding="utf-8")) or {}
    if not isinstance(data, dict):
        return {}
    accounts = data.get("accounts")
    if not isinstance(accounts, dict):
        return {}
    usernames: dict[str, str] = {}
    for account_key, account in accounts.items():
        if not isinstance(account_key, str) or not account_key.startswith("agent_"):
            continue
        if not isinstance(account, dict):
            continue
        username = account.get("username")
        if isinstance(username, str) and username:
            usernames[account_key] = username
    return usernames


_OPTIONAL_DICT_SECTION_NAMES = (
    "teams",
    "cultures",
    "rooms",
    "room_models",
    "room_thread_summary_models",
    "knowledge_bases",
    "mcp_servers",
    "prompts",
    "matrix_room_access",
    "matrix_space",
)
_OPTIONAL_MODEL_SECTION_NAMES = ("debug", "external_trigger_policy", "matrix_sync", "tool_approval")


class ConfigRuntimeValidationError(ValueError):
    """Runtime-aware config validation failed after Pydantic schema validation."""

    def errors(self, *, include_context: bool = False) -> list[dict[str, object]]:
        """Return one ValidationError-like payload for shared config UX code."""
        del include_context
        return [{"loc": ("config",), "msg": str(self), "type": "value_error"}]


CONFIG_LOAD_USER_ERROR_TYPES = (
    ValidationError,
    ConfigRuntimeValidationError,
    yaml.YAMLError,
    OSError,
    UnicodeError,
)


def iter_config_validation_messages(
    exc: ValidationError | ConfigRuntimeValidationError | yaml.YAMLError | OSError | UnicodeError,
) -> list[tuple[str, str]]:
    """Return user-facing validation messages from one config validation exception."""
    if isinstance(exc, ValidationError):
        return [(" → ".join(str(x) for x in error["loc"]), error["msg"]) for error in exc.errors(include_context=False)]
    if isinstance(exc, ConfigRuntimeValidationError):
        return [("config", str(exc))]
    if isinstance(exc, ConfigIncludeError):
        return [("config", str(exc))]
    if isinstance(exc, yaml.YAMLError):
        return [("config", f"Could not parse configuration YAML: {exc}")]
    if isinstance(exc, UnicodeError):
        return [("config", f"Could not read configuration text: {exc}")]
    return [("config", f"Could not load configuration: {exc}")]


def format_invalid_config_message(
    exc: ValidationError | ConfigRuntimeValidationError | yaml.YAMLError | OSError | UnicodeError,
    *,
    footer: str | None = None,
) -> str:
    """Return one shared invalid-configuration message for user-facing surfaces."""
    errors = [f"• {location}: {message}" for location, message in iter_config_validation_messages(exc)]
    response = f"❌ Invalid configuration:\n{'\n'.join(errors)}"
    if footer:
        response = f"{response}\n\n{footer}"
    return response


@dataclass(frozen=True)
class ResolvedRuntimeModel:
    """Resolved active runtime model and context window for one execution context."""

    model_name: str
    context_window: int | None


@dataclass(frozen=True)
class _AuthoredOptionalModel:
    """Static authored semantics for an optional model override field."""

    kind: Literal["unset", "clear", "value"]
    value: str | None = None


@dataclass(frozen=True)
class _StaticCompactionConfigSemantics:
    """Static compaction semantics for one optional model field in one config scope."""

    scope_label: str
    field_name: str
    authored_model: _AuthoredOptionalModel


def _history_policy_from_limits(
    *,
    num_history_runs: int | None,
    num_history_messages: int | None,
) -> HistoryPolicy:
    if num_history_messages is not None:
        return HistoryPolicy(mode="messages", limit=num_history_messages)
    if num_history_runs is not None:
        return HistoryPolicy(mode="runs", limit=num_history_runs)
    return HistoryPolicy(mode="all")


def _normalize_optional_config_sections(data: dict[str, object]) -> None:
    """Replace explicit YAML nulls with the model's expected empty containers."""
    for name in _OPTIONAL_DICT_SECTION_NAMES:
        if data.get(name) is None:
            data[name] = {}
    for name in _OPTIONAL_MODEL_SECTION_NAMES:
        if data.get(name) is None:
            data[name] = {}
    if data.get("plugins") is None:
        data["plugins"] = []


def normalized_config_data(data: object) -> object:
    """Return config input with legacy optional sections normalized."""
    if not isinstance(data, dict):
        return data

    normalized_data = cast("dict[str, object]", data.copy())
    _normalize_optional_config_sections(normalized_data)
    return normalized_data


def _authored_optional_model(model_name: str | None, *, field_is_set: bool) -> _AuthoredOptionalModel:
    """Return the authored tri-state semantics for one optional model field."""
    if not field_is_set:
        return _AuthoredOptionalModel(kind="unset")
    if model_name is None:
        return _AuthoredOptionalModel(kind="clear")
    return _AuthoredOptionalModel(kind="value", value=model_name)


def _strip_empty_root_sections(payload: dict[str, Any]) -> dict[str, Any]:
    """Drop normalized empty root sections from authored config serialization."""
    authored_payload = dict(payload)
    for name in _OPTIONAL_DICT_SECTION_NAMES:
        if authored_payload.get(name) == {}:
            authored_payload.pop(name, None)
    for name in _OPTIONAL_MODEL_SECTION_NAMES:
        if authored_payload.get(name) == {}:
            authored_payload.pop(name, None)
    if authored_payload.get("plugins") == []:
        authored_payload.pop("plugins", None)
    return authored_payload


def _authored_compaction_model_fields(
    compaction: CompactionConfig | CompactionOverrideConfig,
) -> dict[str, _AuthoredOptionalModel]:
    """Return authored tri-state semantics for every optional compaction model field."""
    return {
        "model": _authored_optional_model(
            compaction.model,
            field_is_set="model" in compaction.model_fields_set,
        ),
        "fallback_model": _authored_optional_model(
            compaction.fallback_model,
            field_is_set="fallback_model" in compaction.model_fields_set,
        ),
    }


def _effective_static_compaction_enabled(
    *,
    defaults_enabled: bool,
    override_enabled: bool | None,
    override_fields_set: set[str],
    authored_model_fields: dict[str, _AuthoredOptionalModel],
) -> bool:
    """Resolve whether one authored override block is statically enabled."""
    if "enabled" in override_fields_set:
        return override_enabled is True
    cleared_model_fields = {name for name, authored in authored_model_fields.items() if authored.kind == "clear"}
    if override_fields_set and override_fields_set <= cleared_model_fields:
        return defaults_enabled
    if override_fields_set:
        return True
    return defaults_enabled


def _relative_paths_overlap(left: Path, right: Path) -> bool:
    """Return whether two relative paths overlap by equality, ancestry, or descent."""
    return left == right or left.is_relative_to(right) or right.is_relative_to(left)


@dataclass(frozen=True)
class _KnowledgeBaseSourceSemantics:
    """Source ownership semantics that must match for exact duplicate roots."""

    git_enabled: bool
    git_repo_identity: str
    git_branch: str
    git_credentials_service: str | None
    git_lfs: bool


def _knowledge_base_source_semantics(base_config: KnowledgeBaseConfig) -> _KnowledgeBaseSourceSemantics:
    """Return the source semantics for duplicate-path compatibility checks."""
    git_config = base_config.git
    return _KnowledgeBaseSourceSemantics(
        git_enabled=git_config is not None,
        git_repo_identity=credential_free_repo_url(git_config.repo_url) if git_config is not None else "",
        git_branch=git_config.branch if git_config is not None else "",
        git_credentials_service=git_config.credentials_service if git_config is not None else None,
        git_lfs=git_config.lfs if git_config is not None else False,
    )


def _template_contains_overlapping_subtree(template_dir: Path, target_path: Path) -> bool:
    """Return whether a template already seeds content at or around one target subtree."""
    if not template_dir.is_dir():
        return False
    return any(
        _relative_paths_overlap(source_path.relative_to(template_dir), target_path)
        for source_path in template_dir.rglob("*")
    )


def _skip_private_template_dir_validation(runtime_paths: RuntimePaths | None) -> bool:
    """Return whether runtime-local workers should skip control-plane template validation."""
    if runtime_paths is None:
        return False
    return runtime_paths.env_flag(SANDBOX_RUNTIME_ENV_BY_KEY["runner_mode"]) and bool(
        runtime_paths.env_value(SANDBOX_RUNTIME_ENV_BY_KEY["dedicated_worker_key"], default=""),
    )


def _tool_entry_has_lazy_flag_field(entry: ToolConfigEntry) -> bool:
    """Return whether one normalized tool entry authored a lazy-loading field."""
    return bool(entry.model_fields_set & {"defer", "initial"})


class Config(BaseModel):
    """Complete configuration from YAML."""

    model_config = ConfigDict(extra="forbid")
    _source_files: frozenset[Path] = PrivateAttr(default=frozenset())
    _unavailable_plugin_tool_names: set[str] = PrivateAttr(default_factory=set)
    _unresolved_plugin_tool_sources: frozenset[str] = PrivateAttr(default=frozenset())
    _runtime_approved_egress_injected_default_tool: bool = PrivateAttr(default=False)
    _runtime_approved_egress_injected_approval_rule: bool = PrivateAttr(default=False)
    _runtime_knowledge_base_overlays: dict[str, KnowledgeBaseConfig] = PrivateAttr(default_factory=dict)

    PRIVATE_KNOWLEDGE_BASE_ID_PREFIX: ClassVar[str] = "__agent_private__:"
    TOOL_PRESETS: ClassVar[dict[str, tuple[str, ...]]] = {
        "openclaw_compat": _OPENCLAW_COMPAT_PRESET_TOOLS,
    }
    IMPLIED_TOOLS: ClassVar[dict[str, tuple[str, ...]]] = {
        "matrix_message": ("attachments", "matrix_room"),
    }

    agents: dict[str, AgentConfig] = Field(default_factory=dict, description="Agent configurations")
    teams: dict[str, TeamConfig] = Field(default_factory=dict, description="Team configurations")
    cultures: dict[str, CultureConfig] = Field(default_factory=dict, description="Culture configurations")
    rooms: dict[str, RoomConfig] = Field(default_factory=dict, description="Managed Matrix room metadata")
    room_models: dict[str, str] = Field(default_factory=dict, description="Room-specific model overrides")
    room_thread_summary_models: dict[str, str] = Field(
        default_factory=dict,
        description="Room-specific model overrides for automatic thread summaries",
    )
    plugins: list[PluginEntryConfig] = Field(default_factory=list, description="Plugin entries")
    debug: DebugConfig = Field(default_factory=DebugConfig, description="Debug and diagnostic settings")
    prompts: dict[str, str] = Field(
        default_factory=dict,
        description="Built-in prompt overrides keyed by the uppercase global name from mindroom.prompts",
    )
    defaults: DefaultsConfig = Field(default_factory=DefaultsConfig, description="Default values")
    memory: MemoryConfig = Field(default_factory=MemoryConfig, description="Memory configuration")
    knowledge_bases: dict[str, KnowledgeBaseConfig] = Field(
        default_factory=dict,
        description="Knowledge base configurations keyed by base ID",
    )
    mcp_servers: dict[str, MCPServerConfig] = Field(
        default_factory=dict,
        description="MCP server configurations keyed by server id",
    )
    external_trigger_policy: ExternalTriggerPolicyConfig = Field(
        default_factory=ExternalTriggerPolicyConfig,
        description="Global policy for tool-managed signed external triggers",
    )
    models: dict[str, ModelConfig] = Field(default_factory=dict, description="Model configurations")
    tool_approval: ToolApprovalConfig = Field(
        default_factory=ToolApprovalConfig,
        description="Tool-approval rules for agent-initiated tool calls",
    )
    router: RouterConfig = Field(default_factory=RouterConfig, description="Router configuration")
    voice: VoiceConfig = Field(default_factory=VoiceConfig, description="Voice configuration")
    calls: CallsConfig = Field(default_factory=CallsConfig, description="Voice call (MatrixRTC) configuration")
    cache: CacheConfig = Field(default_factory=CacheConfig, description="Persistent Matrix event cache")
    matrix_sync: MatrixSyncConfig = Field(
        default_factory=MatrixSyncConfig,
        description="Matrix event sync transport configuration",
    )
    timezone: str = Field(
        default="UTC",
        description="Timezone for interpreting scheduling requests and displaying scheduled tasks (e.g., 'America/New_York')",
    )
    mindroom_user: MindRoomUserConfig | None = Field(
        default=None,
        description="Configuration for the internal MindRoom user account (omit for hosted/public profiles)",
    )
    matrix_room_access: MatrixRoomAccessConfig = Field(
        default_factory=MatrixRoomAccessConfig,
        description="Managed Matrix room access/discoverability behavior",
    )
    matrix_space: MatrixSpaceConfig = Field(
        default_factory=MatrixSpaceConfig,
        description="Optional root Matrix Space for grouping managed rooms",
    )
    authorization: AuthorizationConfig = Field(
        default_factory=AuthorizationConfig,
        description="Authorization configuration with fine-grained permissions",
    )
    bot_accounts: list[str] = Field(
        default_factory=list,
        description="Matrix user IDs of non-MindRoom bots (e.g., bridge bots) that should be treated like agents for response logic — their messages won't trigger the multi-human-thread mention requirement",
    )

    @classmethod
    def _lazy_flag_prohibited_message(cls, *, tool_name: str, config_path: str) -> str | None:
        if cls.is_tool_preset(tool_name):
            return (
                f"{config_path}: '{tool_name}' is a preset and cannot be deferred; "
                "defer/initial are only valid on individual tools."
            )
        if tool_name in _DEFER_PROHIBITED_CONTROL_TOOLS:
            return (
                f"{config_path}: '{tool_name}' is a control-plane tool and cannot be deferred; "
                "defer/initial are only valid on runtime tools."
            )
        return None

    @classmethod
    def _validate_raw_tool_lazy_flag_boundary(cls, entry: object, *, config_path: str) -> None:
        name, defer, initial = raw_tool_entry_name_and_lazy_flag_fields(entry)
        if name is None or not (defer or initial):
            return
        if msg := cls._lazy_flag_prohibited_message(tool_name=name, config_path=config_path):
            raise ValueError(msg)

    @model_validator(mode="before")
    @classmethod
    def validate_raw_root_config(cls, data: object) -> object:
        """Normalize optional root sections and reject preset lazy flags before nested validation."""
        normalized = normalized_config_data(data)
        if not isinstance(normalized, dict):
            return normalized

        raw_data = cast("dict[object, object]", normalized)
        for entry in raw_tools_entries(raw_data, "defaults"):
            cls._validate_raw_tool_lazy_flag_boundary(entry, config_path="defaults.tools")

        raw_agents = raw_data.get("agents")
        if not isinstance(raw_agents, dict):
            return normalized

        for agent_name, raw_agent in raw_agents.items():
            if not isinstance(agent_name, str) or not isinstance(raw_agent, dict):
                continue
            agent_data = cast("dict[object, object]", raw_agent)
            tools = agent_data.get("tools")
            if not isinstance(tools, list):
                continue
            for entry in tools:
                cls._validate_raw_tool_lazy_flag_boundary(entry, config_path=f"agents.{agent_name}.tools")

        return normalized

    @model_validator(mode="after")
    def validate_tool_presets_do_not_use_lazy_flags(self) -> Config:
        """Reject lazy-loading control fields on presets after tool-entry coercion."""
        for entry in self.defaults.tools:
            if _tool_entry_has_lazy_flag_field(entry) and (
                msg := self._lazy_flag_prohibited_message(tool_name=entry.name, config_path="defaults.tools")
            ):
                raise ValueError(msg)

        for agent_name, agent_config in self.agents.items():
            for entry in agent_config.tools:
                if _tool_entry_has_lazy_flag_field(entry) and (
                    msg := self._lazy_flag_prohibited_message(
                        tool_name=entry.name,
                        config_path=f"agents.{agent_name}.tools",
                    )
                ):
                    raise ValueError(msg)

        return self

    @field_validator("plugins", mode="before")
    @classmethod
    def normalize_plugins(cls, value: object) -> object:
        """Normalize legacy string plugin entries into structured config objects."""
        if value is None:
            return []
        if not isinstance(value, list):
            return value

        normalized_plugins: list[object] = []
        for plugin_entry in value:
            if isinstance(plugin_entry, str):
                normalized_plugins.append({"path": plugin_entry})
                continue
            normalized_plugins.append(plugin_entry)
        return normalized_plugins

    @field_validator("prompts")
    @classmethod
    def validate_prompt_overrides(cls, value: dict[str, str]) -> dict[str, str]:
        """Ensure prompt overrides map to known built-in string prompt globals."""
        unknown_names = sorted(set(value) - PROMPT_DEFAULT_NAMES)
        if unknown_names:
            allowed = ", ".join(sorted(PROMPT_DEFAULT_NAMES))
            unknown = ", ".join(unknown_names)
            msg = f"Unknown prompt override(s): {unknown}. Allowed prompt names: {allowed}"
            raise ValueError(msg)
        for prompt_name, prompt_text in value.items():
            validate_prompt_template_fields(prompt_name, prompt_text)
        return value

    def get_prompt(self, name: str) -> str:
        """Return one configured prompt override or the built-in default."""
        if name in self.prompts:
            return self.prompts[name]
        return PROMPT_DEFAULTS[name]

    def render_prompt(self, name: str, **kwargs: object) -> str:
        """Render one configured prompt with MindRoom's small bare-field template syntax."""
        return render_prompt_template(self.get_prompt(name), **kwargs)

    @model_validator(mode="after")
    def validate_entity_names(self) -> Config:
        """Ensure agent and team names contain only alphanumeric characters and underscores."""
        invalid_agents = [name for name in self.agents if not _AGENT_NAME_PATTERN.fullmatch(name)]
        invalid_teams = [name for name in self.teams if not _AGENT_NAME_PATTERN.fullmatch(name)]
        invalid_mcp_servers = [name for name in self.mcp_servers if not _AGENT_NAME_PATTERN.fullmatch(name)]
        invalid = sorted(invalid_agents + invalid_teams + invalid_mcp_servers)
        if invalid:
            msg = f"Agent, team, and MCP server names must be alphanumeric/underscore only, got: {', '.join(invalid)}"
            raise ValueError(msg)
        overlapping_names = sorted(set(self.agents) & set(self.teams))
        if overlapping_names:
            msg = f"Agent and team names must be distinct, overlapping keys: {', '.join(overlapping_names)}"
            raise ValueError(msg)
        reserved_entity_names = sorted((set(self.agents) | set(self.teams)) & _RESERVED_ENTITY_NAMES)
        if reserved_entity_names:
            msg = (
                f"Agent and team names must not use reserved internal entity names: {', '.join(reserved_entity_names)}"
            )
            raise ValueError(msg)
        for server_id in self.mcp_servers:
            normalize_mcp_server_id(server_id)
        return self

    @model_validator(mode="after")
    def validate_call_agents(self) -> Config:
        """Ensure call agents and cascaded model references are valid."""
        unknown_agents = sorted(set(self.calls.agents) - set(self.agents))
        if unknown_agents:
            msg = f"calls.agents references unknown agent(s): {', '.join(unknown_agents)}"
            raise ValueError(msg)

        invalid_models = sorted(
            f"{profile_name} -> {profile.model}"
            for profile_name, profile in self.calls.profiles.items()
            if isinstance(profile, CascadedCallProfile)
            and profile.model is not None
            and profile.model not in self.models
        )
        if invalid_models:
            msg = "calls.profiles references unknown cascaded model(s): " + ", ".join(invalid_models)
            raise ValueError(msg)

        agents_by_room: dict[str, list[str]] = {}
        for agent_name in self.calls.agents:
            for room in self.agents[agent_name].rooms:
                agents_by_room.setdefault(room, []).append(agent_name)
        conflicts = [
            f"{room} ({', '.join(sorted(agent_names))})"
            for room, agent_names in sorted(agents_by_room.items())
            if len(agent_names) > 1
        ]
        if conflicts:
            msg = "calls.agents configures multiple agents for room(s): " + "; ".join(conflicts)
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_agent_reply_permissions(self) -> Config:
        """Ensure per-agent reply permissions reference known entities."""
        known_entities = set(self.agents) | set(self.teams) | {ROUTER_AGENT_NAME}
        known_entities.add("*")
        unknown_entities = sorted(set(self.authorization.agent_reply_permissions) - known_entities)
        if unknown_entities:
            msg = f"authorization.agent_reply_permissions contains unknown entities: {', '.join(unknown_entities)}"
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_delegate_to(self) -> Config:
        """Ensure delegate_to targets exist and agents don't delegate to themselves."""
        for agent_name, agent_config in self.agents.items():
            for target in agent_config.delegate_to:
                if target == agent_name:
                    msg = f"Agent '{agent_name}' cannot delegate to itself"
                    raise ValueError(msg)
                if target not in self.agents:
                    msg = f"Agent '{agent_name}' delegates to unknown agent '{target}'"
                    raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_team_agents(self) -> Config:
        """Ensure team members exist and do not use private requester-local state."""
        for team_name, team_config in self.teams.items():
            self.assert_team_agents_supported(team_config.agents, team_name=team_name)
        return self

    def _invalid_compaction_model_references(self) -> list[str]:
        """Return any compaction model references that point at unknown models."""
        invalid_references: list[str] = []
        for semantics in self._static_compaction_semantics():
            if semantics.authored_model.kind != "value":
                continue
            assert semantics.authored_model.value is not None
            if semantics.authored_model.value not in self.models:
                invalid_references.append(
                    f"{semantics.scope_label}.compaction.{semantics.field_name} -> {semantics.authored_model.value}",
                )

        return invalid_references

    def _compaction_models_missing_context_window(self) -> list[str]:
        """Return explicit compaction model references whose target model lacks context_window."""
        invalid_references: list[str] = []
        for semantics in self._static_compaction_semantics():
            if semantics.authored_model.kind != "value":
                continue
            assert semantics.authored_model.value is not None
            if self.models[semantics.authored_model.value].context_window is None:
                invalid_references.append(
                    f"{semantics.scope_label}.compaction.{semantics.field_name} -> {semantics.authored_model.value}",
                )

        return invalid_references

    def _static_compaction_semantics(self) -> list[_StaticCompactionConfigSemantics]:
        """Return static compaction semantics for defaults, agents, and teams."""
        scoped_compactions: list[tuple[str, CompactionConfig | CompactionOverrideConfig]] = []
        defaults_compaction = self.defaults.compaction

        if defaults_compaction is not None:
            scoped_compactions.append(("defaults", defaults_compaction))

        for agent_name, agent_config in self.agents.items():
            if agent_config.compaction is not None:
                scoped_compactions.append((f"agents.{agent_name}", agent_config.compaction))

        for team_name, team_config in self.teams.items():
            if team_config.compaction is not None:
                scoped_compactions.append((f"teams.{team_name}", team_config.compaction))

        return [
            _StaticCompactionConfigSemantics(
                scope_label=scope_label,
                field_name=field_name,
                authored_model=authored_model,
            )
            for scope_label, compaction in scoped_compactions
            for field_name, authored_model in _authored_compaction_model_fields(compaction).items()
        ]

    @model_validator(mode="after")
    def validate_compaction_model_references(self) -> Config:
        """Ensure explicit compaction model and fallback_model references are statically valid."""
        invalid_references = self._invalid_compaction_model_references()
        if invalid_references:
            msg = "Compaction model references unknown models: " + ", ".join(sorted(invalid_references))
            raise ValueError(msg)

        missing_context_windows = self._compaction_models_missing_context_window()
        if missing_context_windows:
            msg = "Explicit compaction model references require a model with context_window: " + ", ".join(
                sorted(missing_context_windows),
            )
            raise ValueError(msg)

        return self

    @model_validator(mode="after")
    def validate_shared_only_integration_assignments(self) -> Config:
        """Reject shared-only integrations on isolating scopes for static and dynamic tool assignments."""
        invalid_assignments: list[str] = []
        for agent_name in sorted(self.agents):
            scope_label = self._agent_scope_label(agent_name)
            execution_scope = self._agent_execution_scope(agent_name)
            unsupported_tools = unsupported_shared_only_integration_names(
                self._get_agent_eager_tools(agent_name),
                execution_scope,
            )
            invalid_assignments.extend(
                f"{agent_name} -> {tool_name} ({scope_label})" for tool_name in unsupported_tools
            )
            for authored_name, incompatible_tools in self._agent_scope_incompatible_deferred_tools(
                agent_name,
            ).items():
                invalid_assignments.extend(
                    f"{agent_name} -> deferred tool '{authored_name}' -> {tool_name} ({scope_label})"
                    for tool_name in incompatible_tools
                )
        if invalid_assignments:
            msg = (
                "Shared-only integrations are supported only for unscoped agents or worker_scope=shared. "
                f"Invalid assignments: {', '.join(invalid_assignments)}"
            )
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_knowledge_base_assignments(self) -> Config:
        """Ensure agents only reference configured knowledge base IDs."""
        invalid_assignments = [
            (agent_name, base_id)
            for agent_name, agent_config in self.agents.items()
            for base_id in agent_config.knowledge_bases
            if base_id not in self.knowledge_bases
        ]
        if invalid_assignments:
            formatted = ", ".join(
                f"{agent_name} -> {base_id}"
                for agent_name, base_id in sorted(invalid_assignments, key=lambda item: (item[0], item[1]))
            )
            msg = f"Agents reference unknown knowledge bases: {formatted}"
            raise ValueError(msg)
        return self

    @classmethod
    def _validate_knowledge_base_ids(cls, base_ids: Collection[str]) -> None:
        """Apply authored knowledge-base ID constraints to any config entry point."""
        reserved_ids = sorted(
            base_id for base_id in base_ids if base_id.startswith(cls.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX)
        )
        if reserved_ids:
            formatted = ", ".join(reserved_ids)
            msg = (
                "knowledge_bases keys must not use the reserved private prefix "
                f"'{cls.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX}'; invalid keys: {formatted}"
            )
            raise ValueError(msg)

        invalid_ids = sorted(base_id for base_id in base_ids if "\n" in base_id or "\r" in base_id)
        if invalid_ids:
            formatted = ", ".join(invalid_ids)
            msg = f"knowledge_bases keys must not contain line breaks; invalid keys: {formatted}"
            raise ValueError(msg)

        invalid_ids = sorted(
            base_id
            for base_id in base_ids
            if not base_id.strip() or base_id in {".", ".."} or "/" in base_id or "\\" in base_id
        )
        if invalid_ids:
            formatted = ", ".join(invalid_ids)
            msg = (
                "knowledge_bases keys must be non-empty single path components without path separators "
                f"or dot segments; invalid keys: {formatted}"
            )
            raise ValueError(msg)

    @model_validator(mode="after")
    def validate_knowledge_base_ids(self) -> Config:
        """Reject reserved, multiline, empty, and path-unsafe knowledge base IDs."""
        self._validate_knowledge_base_ids(self.knowledge_bases)
        return self

    @model_validator(mode="after")
    def validate_knowledge_base_paths_do_not_overlap(self, info: ValidationInfo) -> Config:
        """Reject parent/child top-level knowledge roots while allowing exact aliases."""
        runtime_paths = info.context.get("runtime_paths") if isinstance(info.context, dict) else None
        if runtime_paths is None or len(self.knowledge_bases) < 2:
            return self

        resolved_paths = [
            (base_id, resolve_config_relative_path(base_config.path, runtime_paths).resolve())
            for base_id, base_config in self.knowledge_bases.items()
        ]
        for index, (base_id, root) in enumerate(resolved_paths):
            for other_base_id, other_root in resolved_paths[index + 1 :]:
                if root == other_root:
                    semantics = _knowledge_base_source_semantics(self.knowledge_bases[base_id])
                    other_semantics = _knowledge_base_source_semantics(self.knowledge_bases[other_base_id])
                    if semantics != other_semantics:
                        msg = (
                            "knowledge_bases exact duplicate aliases must use compatible source configuration; "
                            f"'{base_id}' and '{other_base_id}' both resolve to '{root}'"
                        )
                        raise ValueError(msg)
                    continue
                if root.is_relative_to(other_root) or other_root.is_relative_to(root):
                    msg = (
                        "knowledge_bases paths must not overlap unless they are exact duplicate aliases; "
                        f"'{base_id}' resolves to '{root}' and '{other_base_id}' resolves to '{other_root}'"
                    )
                    raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_private_knowledge(self) -> Config:
        """Ensure enabled private knowledge declares an explicit path."""
        invalid_private_knowledge = [
            agent_name
            for agent_name, agent_config in self.agents.items()
            if (
                agent_config.private is not None
                and agent_config.private.knowledge is not None
                and agent_config.private.knowledge.enabled
                and agent_config.private.knowledge.path is None
            )
        ]
        if invalid_private_knowledge:
            formatted = ", ".join(sorted(invalid_private_knowledge))
            msg = f"agents.<name>.private.knowledge.path is required when private.knowledge is enabled; invalid agents: {formatted}"
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_private_git_knowledge_paths(self, info: ValidationInfo) -> Config:
        """Ensure git-backed private knowledge uses a dedicated subtree."""
        memory_notes_dir = Path("memory")
        memory_notes_entrypoint = Path("MEMORY.md")
        runtime_paths = info.context.get("runtime_paths") if isinstance(info.context, dict) else None
        for agent_name, agent_config in self.agents.items():
            private_config = agent_config.private
            if private_config is None or private_config.knowledge is None:
                continue
            private_knowledge = private_config.knowledge
            if private_knowledge.git is None or private_knowledge.path is None:
                continue
            knowledge_path = Path(private_knowledge.path)
            if knowledge_path == Path():
                msg = (
                    f"Agent '{agent_name}' uses git-backed private knowledge at '{private_knowledge.path}', "
                    "but git-backed private knowledge must use a dedicated subtree outside the private root "
                    "and outside scaffolded private workspace content"
                )
                raise ValueError(msg)
            uses_file_memory_backend = self._agent_memory_backend(agent_name) == "file"
            overlaps_private_file_memory = uses_file_memory_backend and _relative_paths_overlap(
                knowledge_path,
                memory_notes_dir,
            )
            if uses_file_memory_backend and _relative_paths_overlap(
                knowledge_path,
                memory_notes_entrypoint,
            ):
                overlaps_private_file_memory = True
            overlaps_template_scaffold = False
            if private_config.template_dir is not None:
                if _relative_paths_overlap(knowledge_path, memory_notes_dir):
                    overlaps_template_scaffold = True
                elif runtime_paths is not None:
                    template_dir = config_relative_path(private_config.template_dir, runtime_paths)
                    overlaps_template_scaffold = _template_contains_overlapping_subtree(template_dir, knowledge_path)
            if overlaps_private_file_memory or overlaps_template_scaffold:
                msg = (
                    f"Agent '{agent_name}' uses git-backed private knowledge at '{private_knowledge.path}', "
                    "but git-backed private knowledge must use a dedicated subtree outside the private root "
                    "and outside scaffolded private workspace content"
                )
                raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_private_template_dirs(self, info: ValidationInfo) -> Config:
        """Ensure private template directories exist when runtime path resolution is available."""
        runtime_paths = info.context.get("runtime_paths") if isinstance(info.context, dict) else None
        if runtime_paths is None or _skip_private_template_dir_validation(runtime_paths):
            return self
        for agent_name, agent_config in self.agents.items():
            private_config = agent_config.private
            if private_config is None or private_config.template_dir is None:
                continue
            template_dir = config_relative_path(private_config.template_dir, runtime_paths)
            try:
                validate_workspace_template_dir(template_dir)
            except ValueError as exc:
                msg = f"Agent '{agent_name}' has invalid private.template_dir: {exc}"
                raise ValueError(msg) from exc
        return self

    @model_validator(mode="after")
    def validate_culture_assignments(self) -> Config:
        """Ensure culture assignments reference known agents and remain one-to-one."""
        unknown_assignments = [
            (culture_name, agent_name)
            for culture_name, culture_config in self.cultures.items()
            for agent_name in culture_config.agents
            if agent_name not in self.agents
        ]
        if unknown_assignments:
            formatted = ", ".join(
                f"{culture_name} -> {agent_name}"
                for culture_name, agent_name in sorted(unknown_assignments, key=lambda item: (item[0], item[1]))
            )
            msg = f"Cultures reference unknown agents: {formatted}"
            raise ValueError(msg)

        agent_to_culture: dict[str, str] = {}
        duplicate_assignments: list[tuple[str, str, str]] = []
        for culture_name, culture_config in self.cultures.items():
            for agent_name in culture_config.agents:
                existing_culture = agent_to_culture.get(agent_name)
                if existing_culture is not None and existing_culture != culture_name:
                    duplicate_assignments.append((agent_name, existing_culture, culture_name))
                    continue
                agent_to_culture[agent_name] = culture_name

        if duplicate_assignments:
            formatted = ", ".join(
                f"{agent_name} -> {culture_a}, {culture_b}"
                for agent_name, culture_a, culture_b in sorted(
                    duplicate_assignments,
                    key=lambda item: (item[0], item[1], item[2]),
                )
            )
            msg = f"Agents cannot belong to multiple cultures: {formatted}"
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_internal_user_username_not_reserved(self, info: ValidationInfo) -> Config:
        """Ensure the internal user localpart does not collide with bot accounts."""
        if self.mindroom_user is None:
            return self
        runtime_paths = info.context.get("runtime_paths") if isinstance(info.context, dict) else None
        if runtime_paths is None:
            return self
        reserved_localparts: dict[str, str] = {}
        persisted_usernames = _persisted_entity_account_usernames(runtime_paths)
        entity_names = [ROUTER_AGENT_NAME, *self.agents, *self.teams]
        for entity_name in entity_names:
            account_key = f"agent_{entity_name}"
            persisted_username = persisted_usernames.get(account_key)
            if persisted_username is None:
                continue
            if entity_name == ROUTER_AGENT_NAME:
                label = f"router '{ROUTER_AGENT_NAME}'"
            elif entity_name in self.agents:
                label = f"agent '{entity_name}'"
            else:
                label = f"team '{entity_name}'"
            reserved_localparts[persisted_username] = label
        conflict = reserved_localparts.get(self.mindroom_user.username)
        if conflict:
            msg = f"mindroom_user.username '{self.mindroom_user.username}' conflicts with {conflict} Matrix localpart"
            raise ValueError(msg)
        return self

    @model_validator(mode="after")
    def validate_root_space_alias_does_not_collide_with_managed_rooms(self, info: ValidationInfo) -> Config:
        """Ensure no managed room key maps to the reserved root Space alias."""
        if not self.matrix_space.enabled:
            return self
        runtime_paths = info.context.get("runtime_paths") if isinstance(info.context, dict) else None
        if runtime_paths is None:
            return self
        reserved_alias_localpart = managed_space_alias_localpart(runtime_paths=runtime_paths)
        colliding_rooms = sorted(
            room_key
            for room_key in self.get_all_configured_rooms()
            if not room_key.startswith(("!", "#"))
            and managed_room_alias_localpart(room_key, runtime_paths=runtime_paths) == reserved_alias_localpart
        )
        if colliding_rooms:
            formatted = ", ".join(colliding_rooms)
            msg = (
                "Managed room keys conflict with the reserved root Space alias "
                f"'{reserved_alias_localpart}': {formatted}"
            )
            raise ValueError(msg)
        return self

    def get_domain(self, runtime_paths: RuntimePaths) -> str:
        """Extract the Matrix domain for one explicit runtime context."""
        homeserver = runtime_matrix_homeserver(runtime_paths)
        return extract_server_name_from_homeserver(homeserver, runtime_paths)

    @property
    def source_files(self) -> frozenset[Path]:
        """Files this config was loaded from: the top-level file plus every include.

        Empty when the config was not loaded from disk via :func:`load_config`.
        """
        return self._source_files

    @classmethod
    def validate_with_runtime(
        cls,
        data: object,
        runtime_paths: RuntimePaths,
        *,
        tolerate_plugin_load_errors: bool = False,
    ) -> Config:
        """Validate config data against one explicit runtime context."""
        normalized_data = normalized_config_data(data)
        approved_egress_overlay = apply_runtime_approved_egress_overlay(normalized_data, runtime_paths)
        config = cls.model_validate(approved_egress_overlay.data, context={"runtime_paths": runtime_paths})
        from mindroom.entity_resolution import validate_call_agent_room_ownership  # noqa: PLC0415

        validate_call_agent_room_ownership(config, runtime_paths)
        config._runtime_approved_egress_injected_default_tool = approved_egress_overlay.injected_default_tool
        config._runtime_approved_egress_injected_approval_rule = approved_egress_overlay.injected_approval_rule
        # why-lazy: module-top catalog import pulls runtime tool registry paths and loads agents+tools at config import.
        from mindroom.tool_system.catalog import ToolConfigOverrideError, ToolMetadataValidationError  # noqa: PLC0415

        try:
            if tolerate_plugin_load_errors:
                config._validate_authored_tool_entries(
                    runtime_paths,
                    tolerate_plugin_load_errors=True,
                )
            else:
                config._validate_authored_tool_entries(runtime_paths)
        except (PluginValidationError, ToolConfigOverrideError, ToolMetadataValidationError) as exc:
            raise ConfigRuntimeValidationError(str(exc)) from exc
        return config

    def authored_model_dump(self) -> dict[str, Any]:
        """Serialize authored config."""
        payload = cast("dict[str, Any]", self.model_dump(exclude_unset=True))
        knowledge_bases = payload.get("knowledge_bases")
        if isinstance(knowledge_bases, dict):
            for base_id in self._runtime_knowledge_base_overlays:
                knowledge_bases.pop(base_id, None)
            if not knowledge_bases:
                payload.pop("knowledge_bases", None)
        payload = strip_runtime_approved_egress_overlay_from_dump(
            payload,
            injected_default_tool=self._runtime_approved_egress_injected_default_tool,
            injected_approval_rule=self._runtime_approved_egress_injected_approval_rule,
        )
        return _strip_empty_root_sections(payload)

    def with_runtime_knowledge_base_overlay(
        self,
        base_id: str,
        base_config: KnowledgeBaseConfig,
    ) -> Config:
        """Return an effective config with one non-authored knowledge base."""
        self._validate_knowledge_base_ids([base_id])
        if base_id in self.knowledge_bases and base_id not in self._runtime_knowledge_base_overlays:
            msg = f"Runtime knowledge base '{base_id}' conflicts with an authored knowledge base"
            raise ValueError(msg)
        config = self.model_copy(
            update={"knowledge_bases": {**self.knowledge_bases, base_id: base_config}},
        )
        config._runtime_knowledge_base_overlays = {
            **self._runtime_knowledge_base_overlays,
            base_id: base_config,
        }
        return config

    def runtime_knowledge_base_overlay(self, base_id: str) -> KnowledgeBaseConfig | None:
        """Return one runtime-only knowledge base overlay, when present."""
        return self._runtime_knowledge_base_overlays.get(base_id)

    def _agent_culture(self, agent_name: str) -> tuple[str, CultureConfig] | None:
        """Get the configured culture assignment for an agent, if any."""
        for culture_name, culture_config in self.cultures.items():
            if agent_name in culture_config.agents:
                return culture_name, culture_config
        return None

    def get_agent(self, agent_name: str) -> AgentConfig:
        """Get an agent configuration by name.

        Args:
            agent_name: Name of the agent

        Returns:
            Agent configuration

        Raises:
            ValueError: If agent not found

        """
        if agent_name not in self.agents:
            available = ", ".join(sorted(self.agents.keys()))
            msg = f"Unknown agent: {agent_name}. Available agents: {available}"
            raise ValueError(msg)
        return self.agents[agent_name]

    def _default_history_settings(self) -> ResolvedHistorySettings:
        """Return defaults-only replay settings for ad hoc shared team scope."""
        return ResolvedHistorySettings(
            policy=_history_policy_from_limits(
                num_history_runs=self.defaults.num_history_runs,
                num_history_messages=self.defaults.num_history_messages,
            ),
            max_tool_calls_from_history=self.defaults.max_tool_calls_from_history,
            system_message_role="system",
        )

    def _entity_history_settings(self, entity_name: str) -> ResolvedHistorySettings:
        """Return effective replay settings for one configured agent or team."""
        if entity_name in self.agents:
            entity = self.get_agent(entity_name)
        elif entity_name in self.teams:
            entity = self.teams[entity_name]
        else:
            msg = f"Unknown entity: {entity_name}"
            raise ValueError(msg)

        num_history_runs = entity.num_history_runs
        num_history_messages = entity.num_history_messages
        if num_history_runs is None and num_history_messages is None:
            num_history_runs = self.defaults.num_history_runs
            num_history_messages = self.defaults.num_history_messages

        max_tool_calls_from_history = (
            entity.max_tool_calls_from_history
            if entity.max_tool_calls_from_history is not None
            else self.defaults.max_tool_calls_from_history
        )
        return ResolvedHistorySettings(
            policy=_history_policy_from_limits(
                num_history_runs=num_history_runs,
                num_history_messages=num_history_messages,
            ),
            max_tool_calls_from_history=max_tool_calls_from_history,
            system_message_role="system",
        )

    def _default_compaction_config(self) -> CompactionConfig:
        """Return the effective destructive compaction config for defaults-only scope."""
        base = self.defaults.compaction
        merged = base.model_dump() if base is not None else {}
        return CompactionConfig.model_validate(merged)

    def _has_authored_default_compaction_config(self) -> bool:
        """Return whether defaults-only scope has authored destructive compaction config."""
        return self.defaults.compaction is not None

    def _entity_compaction_config(self, entity_name: str) -> CompactionConfig:
        """Return the effective destructive compaction config for one configured agent or team."""
        base = self.defaults.compaction
        defaults_enabled = base.enabled if base is not None else False
        merged = base.model_dump() if base is not None else {}
        if entity_name in self.agents:
            override = self.get_agent(entity_name).compaction
        elif entity_name in self.teams:
            override = self.teams[entity_name].compaction
        else:
            msg = f"Unknown entity: {entity_name}"
            raise ValueError(msg)
        if override is not None:
            authored_override = override.model_dump(exclude_unset=True)
            explicit_enabled = authored_override.pop(
                "enabled",
                override.enabled if "enabled" in override.model_fields_set else None,
            )
            for field_name, field_value in authored_override.items():
                if field_value is None:
                    merged.pop(field_name, None)
                    continue
                merged[field_name] = field_value
            if authored_override.get("threshold_tokens") is not None:
                merged.pop("threshold_percent", None)
            if authored_override.get("threshold_percent") is not None:
                merged.pop("threshold_tokens", None)
            merged["enabled"] = _effective_static_compaction_enabled(
                defaults_enabled=defaults_enabled,
                override_enabled=explicit_enabled,
                override_fields_set=override.model_fields_set,
                authored_model_fields=_authored_compaction_model_fields(override),
            )
        return CompactionConfig.model_validate(merged)

    def _has_authored_entity_compaction_config(self, entity_name: str) -> bool:
        """Return whether destructive compaction was explicitly configured for one configured entity."""
        if entity_name in self.agents:
            override = self.get_agent(entity_name).compaction
        elif entity_name in self.teams:
            override = self.teams[entity_name].compaction
        else:
            msg = f"Unknown entity: {entity_name}"
            raise ValueError(msg)
        return self.defaults.compaction is not None or override is not None

    def resolve_entity(self, entity_name: str | None) -> ResolvedEntityView:
        """Return the resolved config view for one entity, or the defaults-only scope for None."""
        return ResolvedEntityView(_config=self, name=entity_name)

    def get_model_context_window(self, model_name: str) -> int | None:
        """Return the configured context window for one model name, when known."""
        model_config = self.models.get(model_name)
        return model_config.context_window if model_config and model_config.context_window else None

    def _deferred_tool_scope_incompatible_tools(
        self,
        agent_name: str,
        authored_tool_name: str,
    ) -> list[str]:
        """Return expanded deferred tools invalid for one agent's effective execution scope."""
        authored_config = self._agent_authored_deferred_tool_config(agent_name, authored_tool_name)
        if authored_config is None:
            return []
        execution_scope = self._agent_execution_scope(agent_name)
        return unsupported_shared_only_integration_names(
            self.expand_tool_names([authored_tool_name]),
            execution_scope,
        )

    def _agent_scope_incompatible_deferred_tools(self, agent_name: str) -> dict[str, list[str]]:
        """Return deferred authored tools whose expanded contents are invalid for one agent scope."""
        return {
            entry.name: incompatible_tools
            for entry in self._agent_authored_deferred_tool_configs(agent_name)
            if (incompatible_tools := self._deferred_tool_scope_incompatible_tools(agent_name, entry.name))
        }

    def get_worker_grantable_credentials(self) -> frozenset[str]:
        """Return shared credential service names allowed inside isolated workers."""
        configured = self.defaults.worker_grantable_credentials
        if configured is None:
            return DEFAULT_WORKER_GRANTABLE_CREDENTIALS
        return frozenset(configured)

    def _agent_execution_scope(self, agent_name: str) -> WorkerScope | None:
        """Return the internal derived execution scope for one agent.

        This is not the authored config field.
        Shared agents derive it from `worker_scope` (or defaults), while private agents
        derive the same runtime concept from `private.per`.
        """
        policy = resolve_agent_policy_from_data(
            agent_name,
            self.get_agent(agent_name),
            default_worker_scope=self.defaults.worker_scope,
            private_knowledge_base_id_prefix=self.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX,
        )
        return policy.effective_execution_scope

    def _agent_scope_label(self, agent_name: str) -> str:
        """Return the user-facing authored scope label for one agent.

        Keep this separate from `_agent_execution_scope()`: the internal runtime uses
        one derived execution scope, but user-facing messages should still distinguish
        authored `worker_scope=...` from private `private.per=...`.
        """
        policy = resolve_agent_policy_from_data(
            agent_name,
            self.get_agent(agent_name),
            default_worker_scope=self.defaults.worker_scope,
            private_knowledge_base_id_prefix=self.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX,
        )
        return policy.scope_label

    def _get_agent_eager_tools(self, agent_name: str) -> list[str]:
        """Return expanded non-deferred tools visible without a dynamic load."""
        tool_names: list[str] = []
        for entry in self._get_agent_authored_tool_configs(agent_name):
            if entry.defer:
                continue
            tool_names.extend(self.expand_tool_names([entry.name]))
        return tool_names

    def _agent_private_knowledge_base_id(self, agent_name: str) -> str | None:
        """Return the synthetic knowledge base ID for one agent's private knowledge."""
        policy = resolve_agent_policy_from_data(
            agent_name,
            self.get_agent(agent_name),
            default_worker_scope=self.defaults.worker_scope,
            private_knowledge_base_id_prefix=self.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX,
        )
        return policy.private_knowledge_base_id

    def get_private_knowledge_base_agent(self, base_id: str) -> str | None:
        """Return the owning agent for a synthetic private knowledge base ID."""
        return resolve_private_knowledge_base_agent(
            base_id,
            build_agent_policy_seeds(
                self.agents,
                default_worker_scope=self.defaults.worker_scope,
            ),
            private_knowledge_base_id_prefix=self.PRIVATE_KNOWLEDGE_BASE_ID_PREFIX,
        )

    def _agent_knowledge_base_ids(self, agent_name: str) -> list[str]:
        """Return shared and private knowledge base IDs assigned to one agent."""
        agent_config = self.get_agent(agent_name)
        base_ids = list(agent_config.knowledge_bases)
        private_base_id = self._agent_private_knowledge_base_id(agent_name)
        if private_base_id is not None:
            base_ids.append(private_base_id)
        return base_ids

    def get_knowledge_base_config(self, base_id: str) -> KnowledgeBaseConfig:
        """Return one effective knowledge base config, including synthetic private bases."""
        configured = self.knowledge_bases.get(base_id)
        if configured is not None:
            return configured

        agent_name = self.get_private_knowledge_base_agent(base_id)
        if agent_name is None:
            msg = f"Knowledge base '{base_id}' is not configured"
            raise ValueError(msg)

        agent_config = self.get_agent(agent_name)
        private_config = agent_config.private
        if private_config is None:
            msg = f"Knowledge base '{base_id}' is not configured"
            raise ValueError(msg)

        private_knowledge = private_config.knowledge
        if private_knowledge is None or not private_knowledge.enabled:
            msg = f"Knowledge base '{base_id}' is not configured"
            raise ValueError(msg)

        knowledge_path = private_knowledge.path
        if knowledge_path is None:
            msg = f"Knowledge base '{base_id}' is not configured"
            raise ValueError(msg)

        return KnowledgeBaseConfig(
            description=private_knowledge.description,
            path=knowledge_path,
            watch=private_knowledge.watch,
            chunk_size=private_knowledge.chunk_size,
            chunk_overlap=private_knowledge.chunk_overlap,
            git=private_knowledge.git,
        )

    def _validate_authored_tool_entry(
        self,
        entry: ToolConfigEntry,
        *,
        config_path_prefix: str,
        tool_validation_snapshot: dict[str, ToolValidationInfo],
    ) -> None:
        """Validate one authored tool entry against the resolved validation snapshot."""
        # why-lazy: module-top catalog import pulls runtime tool registry paths and loads agents+tools at config import.
        from mindroom.tool_system.catalog import (  # noqa: PLC0415
            ToolConfigOverrideError,
            validate_authored_tool_entry_overrides,
        )

        validation_info = tool_validation_snapshot.get(entry.name)
        if entry.name not in tool_validation_snapshot and not self.is_tool_preset(entry.name):
            if self._unresolved_plugin_tool_sources:
                logger.warning(
                    "Unknown tool may belong to a plugin whose tool names could not be resolved; "
                    "disabling it for this run (verify the tool name is not a typo)",
                    config_path=config_path_prefix,
                    tool_name=entry.name,
                    unresolved_plugin_sources=sorted(self._unresolved_plugin_tool_sources),
                )
                self._unavailable_plugin_tool_names.add(entry.name)
                return
            msg = f"{config_path_prefix}.{entry.name}: Unknown tool '{entry.name}'."
            raise ToolConfigOverrideError(msg)
        if validation_info is not None and validation_info.unavailable_due_to_plugin_load_error:
            logger.warning(
                "Plugin tool unavailable because plugin failed to load",
                config_path=config_path_prefix,
                tool_name=entry.name,
            )
            return

        validate_authored_tool_entry_overrides(
            entry.name,
            entry.overrides,
            config_path_prefix=config_path_prefix,
            tool_metadata=tool_validation_snapshot,
        )

    def _validate_authored_tool_entries(
        self,
        runtime_paths: RuntimePaths,
        *,
        tolerate_plugin_load_errors: bool = False,
    ) -> None:
        """Validate authored tool references against one resolved validation snapshot."""
        # why-lazy: module-top catalog import pulls runtime tool registry paths and loads agents+tools at config import.
        from mindroom.tool_system.catalog import (  # noqa: PLC0415
            resolved_tool_validation_snapshot_for_runtime,
            unresolved_plugin_tool_sources_for_runtime,
        )

        tool_validation_snapshot = resolved_tool_validation_snapshot_for_runtime(
            runtime_paths,
            self,
            tolerate_plugin_load_errors=tolerate_plugin_load_errors,
        )
        self._unresolved_plugin_tool_sources = unresolved_plugin_tool_sources_for_runtime(
            runtime_paths,
            self,
            tolerate_plugin_load_errors=tolerate_plugin_load_errors,
        )
        self._unavailable_plugin_tool_names = {
            tool_name
            for tool_name, validation_info in tool_validation_snapshot.items()
            if validation_info.unavailable_due_to_plugin_load_error
        }
        self._validate_authored_tool_entries_with_snapshot(
            tool_validation_snapshot=tool_validation_snapshot,
        )

    def _validate_authored_tool_entries_with_snapshot(
        self,
        *,
        tool_validation_snapshot: dict[str, ToolValidationInfo],
    ) -> None:
        """Validate authored tool references against one already-resolved validation snapshot."""
        for index, entry in enumerate(self.defaults.tools):
            self._validate_authored_tool_entry(
                entry,
                config_path_prefix=f"defaults.tools[{index}]",
                tool_validation_snapshot=tool_validation_snapshot,
            )
        for agent_name, agent_config in self.agents.items():
            for index, entry in enumerate(agent_config.tools):
                self._validate_authored_tool_entry(
                    entry,
                    config_path_prefix=f"agents.{agent_name}.tools[{index}]",
                    tool_validation_snapshot=tool_validation_snapshot,
                )

    def _get_agent_authored_tool_configs(self, agent_name: str) -> list[EffectiveToolConfig]:
        """Return effective authored tool config entries before preset/implied expansion."""
        from mindroom.tool_system.catalog import apply_authored_overrides  # noqa: PLC0415

        agent_config = self.get_agent(agent_name)
        default_entries_by_name = {entry.name: entry for entry in self.defaults.tools}
        agent_entry_names = {entry.name for entry in agent_config.tools}
        effective_authored_entries: list[EffectiveToolConfig] = []

        if agent_config.include_default_tools:
            for entry in agent_config.tools:
                if not self._tool_name_is_available(entry.name):
                    continue
                base_overrides: dict[str, object] = {}
                if default_entry := default_entries_by_name.get(entry.name):
                    base_overrides = apply_authored_overrides({}, default_entry.overrides)
                effective_authored_entries.append(
                    EffectiveToolConfig(
                        name=entry.name,
                        tool_config_overrides=apply_authored_overrides(base_overrides, entry.overrides),
                        defer=entry.defer,
                        initial=entry.initial,
                        authored_order=len(effective_authored_entries),
                        authored_name=entry.name,
                    ),
                )
            for entry in self.defaults.tools:
                if entry.name in agent_entry_names:
                    continue
                if not self._tool_name_is_available(entry.name):
                    continue
                effective_authored_entries.append(
                    EffectiveToolConfig(
                        name=entry.name,
                        tool_config_overrides=apply_authored_overrides({}, entry.overrides),
                        authored_order=len(effective_authored_entries),
                        authored_name=entry.name,
                    ),
                )
        else:
            for entry in agent_config.tools:
                if not self._tool_name_is_available(entry.name):
                    continue
                effective_authored_entries.append(
                    EffectiveToolConfig(
                        name=entry.name,
                        tool_config_overrides=apply_authored_overrides({}, entry.overrides),
                        defer=entry.defer,
                        initial=entry.initial,
                        authored_order=len(effective_authored_entries),
                        authored_name=entry.name,
                    ),
                )

        return effective_authored_entries

    def _tool_name_is_available(self, tool_name: str) -> bool:
        """Return whether an authored tool survived tolerant plugin-load validation."""
        return tool_name not in self._unavailable_plugin_tool_names

    def _agent_tool_configs(self, agent_name: str) -> list[EffectiveToolConfig]:
        """Return effective runtime tool config entries for each authored owner."""
        effective_entries = []
        for authored_entry in self._get_agent_authored_tool_configs(agent_name):
            if not self._tool_name_is_available(authored_entry.name):
                continue
            effective_entries.extend(
                (
                    EffectiveToolConfig(
                        name=tool_name,
                        tool_config_overrides=(
                            dict(authored_entry.tool_config_overrides) if tool_name == authored_entry.name else {}
                        ),
                        defer=authored_entry.defer,
                        initial=authored_entry.initial,
                        authored_order=authored_entry.authored_order,
                        authored_name=authored_entry.name,
                    )
                )
                for tool_name in self.expand_tool_names([authored_entry.name])
                if tool_name not in self._unavailable_plugin_tool_names
            )
        return effective_entries

    def _agent_available_tools(self, agent_name: str) -> list[str]:
        """Get all tools the agent may use after dynamic loading."""
        agent_config = self.get_agent(agent_name)
        explicit_names = [name for name in agent_config.tool_names if self._tool_name_is_available(name)]
        if agent_config.include_default_tools:
            explicit_names.extend(name for name in self.defaults.tool_names if self._tool_name_is_available(name))
        return self.expand_tool_names(explicit_names)

    def _agent_authored_deferred_tool_configs(self, agent_name: str) -> list[EffectiveToolConfig]:
        """Return one entry per authored deferred tool in effective order."""
        return [
            EffectiveToolConfig(
                name=entry.name,
                tool_config_overrides=dict(entry.tool_config_overrides),
                defer=entry.defer,
                initial=entry.initial,
                authored_order=entry.authored_order,
                authored_name=entry.name,
            )
            for entry in self._get_agent_authored_tool_configs(agent_name)
            if entry.defer and self._tool_name_is_available(entry.name)
        ]

    def _agent_authored_deferred_tool_config(
        self,
        agent_name: str,
        authored_tool_name: str,
    ) -> EffectiveToolConfig | None:
        """Return one authored deferred tool config by authored name."""
        for entry in self._agent_authored_deferred_tool_configs(agent_name):
            if entry.name == authored_tool_name:
                return entry
        return None

    def _agent_tool_runtime_overrides(self, agent_name: str, tool_name: str) -> dict[str, object] | None:
        """Return runtime kwargs derived from one agent's authored tool overrides."""
        from mindroom.tool_system.catalog import authored_tool_overrides_to_runtime  # noqa: PLC0415

        agent_config = self.get_agent(agent_name)
        overrides = agent_config.get_tool_overrides(tool_name)
        if not overrides:
            return None

        return authored_tool_overrides_to_runtime(tool_name, overrides)

    def _agent_hard_dependency_tool_names(self, agent_name: str) -> set[str]:
        """Return tool names that are hard startup dependencies for one agent."""
        referenced_tool_names: set[str] = set()
        for entry in self._agent_tool_configs(agent_name):
            if not entry.defer or entry.initial:
                referenced_tool_names.add(entry.name)
        return referenced_tool_names

    def get_entities_referencing_tools(self, tool_names: set[str]) -> set[str]:
        """Return agents and teams that depend on any of the given tools."""
        matching_agents = {
            agent_name for agent_name in self.agents if self._agent_hard_dependency_tool_names(agent_name) & tool_names
        }
        matching_teams = {
            team_name
            for team_name, team_config in self.teams.items()
            if any(agent_name in matching_agents for agent_name in team_config.agents)
        }
        return matching_agents | matching_teams

    def get_agent_delegation_closure(
        self,
        agent_name: str,
        *,
        closures: dict[str, frozenset[str]] | None = None,
    ) -> frozenset[str]:
        """Return one agent plus all agents reachable through transitive delegation."""
        return get_agent_delegation_closure(
            agent_name,
            build_agent_policy_seeds(
                self.agents,
                default_worker_scope=self.defaults.worker_scope,
            ),
            closures=closures,
        )

    def get_unsupported_team_agents(
        self,
        agent_names: list[str],
        *,
        closures: dict[str, frozenset[str]] | None = None,
        allow_direct_private_agents: bool = False,
    ) -> dict[str, tuple[str, ...] | None]:
        """Return unsupported team members keyed by agent name.

        Unknown agents map to `None`.
        Supported known agents are omitted.
        Private or transitively private members map to their reachable private targets.
        """
        return get_unsupported_team_agents(
            agent_names,
            build_agent_policy_seeds(
                self.agents,
                default_worker_scope=self.defaults.worker_scope,
            ),
            closures=closures,
            allow_direct_private_agents=allow_direct_private_agents,
        )

    @staticmethod
    def unsupported_team_agent_message(
        agent_name: str,
        *,
        prefix: str,
        private_targets: tuple[str, ...] | None,
    ) -> str:
        """Return the user-facing error for one unsupported team member."""
        return unsupported_team_agent_message(
            agent_name,
            prefix=prefix,
            private_targets=private_targets,
        )

    def assert_team_agents_supported(
        self,
        agent_names: list[str],
        *,
        team_name: str | None = None,
        allow_direct_private_agents: bool = False,
    ) -> None:
        """Reject unknown or currently unsupported team members."""
        prefix = f"Team '{team_name}'" if team_name is not None else "Team request"
        closure_cache: dict[str, frozenset[str]] = {}
        unsupported_agents = self.get_unsupported_team_agents(
            agent_names,
            closures=closure_cache,
            allow_direct_private_agents=allow_direct_private_agents,
        )
        if not unsupported_agents:
            return
        first_unsupported_agent, private_targets = next(iter(unsupported_agents.items()))
        raise ValueError(
            self.unsupported_team_agent_message(
                first_unsupported_agent,
                prefix=prefix,
                private_targets=private_targets,
            ),
        )

    @classmethod
    def get_tool_preset(cls, tool_name: str) -> tuple[str, ...] | None:
        """Return the tool expansion for a preset name."""
        return cls.TOOL_PRESETS.get(tool_name)

    @classmethod
    def is_tool_preset(cls, tool_name: str) -> bool:
        """Return whether a tool name is a known config preset."""
        return tool_name in cls.TOOL_PRESETS

    @classmethod
    def expand_tool_names(cls, tool_names: list[str]) -> list[str]:
        """Expand tool presets and implied tools, deduping while preserving order."""
        expanded: list[str] = []
        seen: set[str] = set()
        queue = deque(tool_names)
        while queue:
            tool_name = queue.popleft()
            if tool_name in seen:
                continue
            seen.add(tool_name)
            expanded.append(tool_name)
            next_tools = list(cls.get_tool_preset(tool_name) or ())
            next_tools.extend(cls.IMPLIED_TOOLS.get(tool_name, ()))
            queue.extend(implied_tool for implied_tool in next_tools if implied_tool not in seen)
        return expanded

    def _agent_memory_backend(self, agent_name: str) -> MemoryBackend:
        """Get effective memory backend for one agent."""
        agent_config = self.agents.get(agent_name)
        if agent_config is None:
            return self.memory.backend
        if agent_config.memory_backend is not None:
            return agent_config.memory_backend
        return self.memory.backend

    def _agent_memory_search(self, agent_name: str) -> MemorySearchConfig:
        """Get effective file-memory search settings for one agent."""
        agent_config = self.agents.get(agent_name)
        override = agent_config.memory_search if agent_config is not None else None
        if override is None:
            return self.memory.search
        # exclude_none keeps the "None inherits" tri-state; deep copy avoids aliasing memory.search.include.
        return self.memory.search.model_copy(update=override.model_dump(exclude_none=True), deep=True)

    def uses_file_memory(self) -> bool:
        """Return whether any configured agent uses file-backed memory."""
        if not self.agents:
            return self.memory.backend == "file"
        return any(self._agent_memory_backend(agent_name) == "file" for agent_name in self.agents)

    def get_all_configured_rooms(self) -> set[str]:
        """Extract all configured room references.

        Returns:
            Set of all unique room references from room, agent, and team configurations

        """
        all_room_aliases = set(self.rooms)
        for agent_config in self.agents.values():
            all_room_aliases.update(agent_config.rooms)
        for team_config in self.teams.values():
            all_room_aliases.update(team_config.rooms)
        return all_room_aliases

    def get_entity_thread_mode(
        self,
        entity_name: str,
        runtime_paths: RuntimePaths,
        room_id: str | None = None,
    ) -> Literal["thread", "room"]:
        """Get effective thread mode for an agent, team, or router.

        Agents use their explicit per-agent setting.
        Teams inherit a mode only when all member agents share it.
        Router inherits a mode only when all relevant configured agents share it.
        In ambiguous cases, default to "thread".
        """
        from mindroom.entity_resolution import resolve_agent_thread_mode, router_agents_for_room  # noqa: PLC0415

        runtime_room_override = resolve_room_thread_mode_override(runtime_paths, room_id)
        if runtime_room_override is not None:
            return runtime_room_override

        if entity_name in self.agents:
            return resolve_agent_thread_mode(
                self.agents[entity_name],
                room_id,
                runtime_paths,
            )

        if entity_name in self.teams:
            team_modes: set[Literal["thread", "room"]] = {
                resolve_agent_thread_mode(
                    self.agents[name],
                    room_id,
                    runtime_paths,
                )
                for name in self.teams[entity_name].agents
                if name in self.agents
            }
            if len(team_modes) == 1:
                return next(iter(team_modes))

        if entity_name == ROUTER_AGENT_NAME:
            router_agents = router_agents_for_room(
                self.agents,
                self.teams,
                room_id,
                runtime_paths,
            )
            configured_modes: set[Literal["thread", "room"]] = {
                resolve_agent_thread_mode(
                    self.agents[agent_name],
                    room_id,
                    runtime_paths,
                )
                for agent_name in router_agents
            }
            if len(configured_modes) == 1:
                return next(iter(configured_modes))

        return "thread"

    def _entity_model_name(self, entity_name: str) -> str:
        """Get the model name for an agent, team, or router.

        Args:
            entity_name: Name of the entity (agent, team, or router)

        Returns:
            Model alias (e.g., "default", "fast", etc.)

        Raises:
            ValueError: If entity_name is not found in configuration

        """
        # Router uses router model
        if entity_name == ROUTER_AGENT_NAME:
            return self.router.model
        # Teams use their configured model (required to have one)
        if entity_name in self.teams:
            model = self.teams[entity_name].model
            if model is None:
                msg = f"Team {entity_name} has no model configured"
                raise ValueError(msg)
            return model
        # Regular agents use their configured model
        if entity_name in self.agents:
            return self.agents[entity_name].model

        # Entity not found in any category
        available = sorted(set(self.agents.keys()) | set(self.teams.keys()) | {ROUTER_AGENT_NAME})
        msg = f"Unknown entity: {entity_name}. Available entities: {', '.join(available)}"
        raise ValueError(msg)

    def resolve_runtime_model(
        self,
        *,
        entity_name: str | None,
        active_model_name: str | None = None,
        active_context_window: int | None = None,
        room_id: str | None = None,
        thread_id: str | None = None,
        runtime_paths: RuntimePaths | None = None,
        default_model_name: str = "default",
    ) -> ResolvedRuntimeModel:
        """Resolve the active runtime model plus its configured context window.

        Precedence: explicit `active_model_name`, then a persisted per-thread
        override, then the room override, then the entity's authored model.
        """
        resolved_model_name = active_model_name
        if resolved_model_name is None and thread_id is not None:
            if runtime_paths is None:
                msg = "runtime_paths are required to resolve a thread-specific runtime model"
                raise ValueError(msg)
            thread_override = resolve_thread_model_override(
                runtime_paths,
                thread_id,
                configured_models=self.models,
            ).active
            if thread_override is not None:
                resolved_model_name = thread_override
        if resolved_model_name is None:
            if entity_name is None:
                resolved_model_name = default_model_name
            elif room_id is not None:
                if runtime_paths is None:
                    msg = "runtime_paths are required to resolve a room-specific runtime model"
                    raise ValueError(msg)
                from mindroom.entity_resolution import effective_entity_model_name  # noqa: PLC0415

                resolved_model_name = effective_entity_model_name(self, entity_name, room_id, runtime_paths)
            else:
                resolved_model_name = self._entity_model_name(entity_name)

        resolved_context_window = active_context_window
        if resolved_context_window is None:
            resolved_context_window = self.get_model_context_window(resolved_model_name)

        return ResolvedRuntimeModel(model_name=resolved_model_name, context_window=resolved_context_window)


def load_config(
    runtime_paths: RuntimePaths,
    *,
    tolerate_plugin_load_errors: bool = False,
) -> Config:
    """Load and validate one config against an explicit runtime context."""
    path = runtime_paths.config_path
    if not path.exists():
        msg = f"Agent configuration file not found: {path}"
        raise FileNotFoundError(msg)

    data, source_files = load_yaml_config_source(path)

    try:
        config = Config.validate_with_runtime(
            data,
            runtime_paths,
            tolerate_plugin_load_errors=tolerate_plugin_load_errors,
        )
    except CONFIG_LOAD_USER_ERROR_TYPES as exc:
        # Parsing succeeded, so the full file set is known; expose it the same
        # way as parse-time failures so reload watchers keep covering it.
        attach_partial_source_files(exc, source_files)
        raise
    config._source_files = source_files
    logger.info("loaded_agent_configuration", path=str(path), source_file_count=len(source_files))
    logger.info("loaded_agent_configuration_count", agent_count=len(config.agents))
    return config


def load_config_or_user_error(
    runtime_paths: RuntimePaths,
    *,
    footer: str | None = None,
    tolerate_plugin_load_errors: bool = False,
) -> tuple[Config | None, str | None]:
    """Load config or return one shared user-facing invalid-configuration message."""
    try:
        return load_config(
            runtime_paths,
            tolerate_plugin_load_errors=tolerate_plugin_load_errors,
        ), None
    except CONFIG_LOAD_USER_ERROR_TYPES as exc:
        return None, format_invalid_config_message(exc, footer=footer)
