"""Agent loader that reads agent configurations from YAML file."""

from __future__ import annotations

from dataclasses import dataclass, replace
from datetime import datetime
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from weakref import WeakValueDictionary
from zoneinfo import ZoneInfo

from agno.culture.manager import CultureManager
from agno.db.base import BaseDb, SessionType
from agno.knowledge.knowledge import Knowledge
from agno.learn import LearningMachine, LearningMode, UserMemoryConfig, UserProfileConfig
from agno.run.agent import RunOutput
from agno.run.team import TeamRunOutput

import mindroom.tools  # noqa: F401
from mindroom import agent_storage, constants, model_loading
from mindroom.agent_descriptions import describe_agent
from mindroom.agent_knowledge_descriptions import KnowledgeToolDescribingAgent as Agent
from mindroom.agent_knowledge_descriptions import knowledge_source_descriptions
from mindroom.claude_prompt_cache import install_claude_deferred_tool_search, native_tool_search_supported
from mindroom.credentials import get_runtime_credentials_manager
from mindroom.entity_resolution import entity_identity_registry
from mindroom.hooks import HookRegistry
from mindroom.logging_config import get_logger
from mindroom.openai_tool_search import install_openai_deferred_tool_search, openai_native_tool_search_supported
from mindroom.prompt_templates import build_agent_identity_context, render_prompt_template
from mindroom.runtime_resolution import (
    ResolvedAgentRuntime,
    resolve_agent_runtime,
    resolve_private_requester_scope_root,
)
from mindroom.timing import timed, timed_block
from mindroom.tool_approval import tool_requires_approval_for_openai_compat
from mindroom.tool_system.catalog import (
    TOOL_METADATA,
    default_worker_routed_tools,
    ensure_tool_registry_loaded,
    get_tool_by_name,
)
from mindroom.tool_system.dynamic_toolkits import (
    VisibleToolSurface,
    deferred_tool_catalog_entries,
    has_deferred_tools,
    resolve_dynamic_tool_selection,
    suppress_fully_deferred_toolkit_instructions,
    visible_tool_surface,
)
from mindroom.tool_system.output_files import ToolOutputFilePolicy, wrap_toolkit_for_output_files
from mindroom.tool_system.plugins import load_plugins
from mindroom.tool_system.runtime_context import ToolDispatchContext
from mindroom.tool_system.sandbox_proxy import sandbox_proxy_enabled_for_tool
from mindroom.tool_system.skills import build_agent_skills
from mindroom.tool_system.tool_hooks import build_tool_hook_bridge, prepend_tool_hook_bridge
from mindroom.tool_system.worker_routing import (
    agent_workspace_root_path,
    build_agent_toolkit_worker_target,
    resolve_agent_owned_path,
    shared_storage_root,
)
from mindroom.workers.runtime import primary_worker_backend_name
from mindroom.workspaces import ensure_workspace_template

if TYPE_CHECKING:
    from collections.abc import Callable
    from contextlib import AbstractContextManager

    from agno.knowledge.protocol import KnowledgeProtocol
    from agno.models.base import Model
    from agno.skills import Skills
    from agno.team import Team
    from agno.tools.function import Function
    from agno.tools.toolkit import Toolkit

    from mindroom.agent_knowledge_descriptions import KnowledgeSourceDescription
    from mindroom.config.agent import AgentConfig, CultureConfig, CultureMode
    from mindroom.config.main import Config
    from mindroom.config.models import DefaultsConfig
    from mindroom.credentials import CredentialsManager
    from mindroom.hooks import HookRegistryPlugin
    from mindroom.knowledge.refresh_scheduler import KnowledgeRefreshScheduler
    from mindroom.tool_system.worker_routing import ToolExecutionIdentity, WorkerScope

logger = get_logger(__name__)

_DEFAULT_MIND_AGENT_NAME = "mind"
_DEFAULT_MIND_CONTEXT_FILES = (
    "SOUL.md",
    "AGENTS.md",
    "USER.md",
    "IDENTITY.md",
    "TOOLS.md",
    "HEARTBEAT.md",
)
_PROJECTED_WORKER_ASSET_PATH_PREFIXES = (
    "./.mindroom-worker-assets/",
    ".mindroom-worker-assets/",
)


@dataclass
class _CachedCultureManager:
    """Cached culture manager with a signature for invalidation on config changes."""

    signature: tuple[str, str]
    manager: CultureManager


@dataclass(frozen=True)
class _CultureAgentSettings:
    """Culture feature flags to apply to the Agent constructor."""

    add_culture_to_context: bool
    update_cultural_knowledge: bool
    enable_agentic_culture: bool


@dataclass
class _AdditionalContextChunk:
    """Chunk of preload context with omission metadata.

    ``omitted_chars`` records what the preload cap removed from this chunk so
    the rendered section can tell the model which file was cut and by how much,
    instead of letting a dropped file vanish without a trace.
    """

    title: str
    body: str
    omitted_chars: int = 0


@dataclass(frozen=True)
class _AgentToolAssembly:
    """Toolkits and dynamic-tool visibility assembled for one agent instance."""

    tools: list[Toolkit]
    loaded_tools: tuple[str, ...]
    hidden_toolkits: frozenset[str]
    selected_dynamic_tools: tuple[str, ...]
    # Authored toolkit names whose schemas use native deferred loading.
    deferred_tool_names: tuple[str, ...]
    # Wire-level function names to send with defer_loading on the native
    # server-side tool-search path (Anthropic and OpenAI Responses); empty on
    # the homegrown dynamic-tools path.
    deferred_wire_tool_names: frozenset[str]
    local_tool_names: tuple[str, ...]
    worker_routed_tool_names: tuple[str, ...]


@dataclass(frozen=True)
class _AgentRoleContext:
    """Resolved model name and fully rendered role prompt for one agent instance."""

    model_name: str
    role: str


@dataclass(frozen=True)
class _AgentCultureState:
    """Culture manager and Agent constructor culture flags for one agent instance."""

    manager: CultureManager | None
    add_culture_to_context: bool | None
    update_cultural_knowledge: bool
    enable_agentic_culture: bool


_CULTURE_MANAGER_CACHE: dict[tuple[str, str], _CachedCultureManager] = {}
_PRIVATE_CULTURE_MANAGER_CACHE: WeakValueDictionary[
    tuple[str, str, tuple[str, str]],
    CultureManager,
] = WeakValueDictionary()


def show_tool_calls_for_agent(config: Config, agent_name: str) -> bool:
    """Resolve tool-call visibility for one agent from current config."""
    agent_config = config.agents.get(agent_name)
    if agent_config and agent_config.show_tool_calls is not None:
        return agent_config.show_tool_calls
    return config.defaults.show_tool_calls


def _uses_default_mind_workspace_scaffold(agent_name: str, agent_config: AgentConfig) -> bool:
    return (
        agent_name == _DEFAULT_MIND_AGENT_NAME
        and agent_config.private is None
        and agent_config.memory_backend == "file"
        and tuple(agent_config.context_files) == _DEFAULT_MIND_CONTEXT_FILES
    )


def _ensure_default_mind_workspace(storage_path: Path) -> None:
    workspace_path = agent_workspace_root_path(storage_path, _DEFAULT_MIND_AGENT_NAME)
    ensure_workspace_template(workspace_path, template="mind")


def ensure_default_agent_workspaces(config: Config, storage_path: Path) -> None:
    """Materialize built-in starter workspaces under the active runtime storage root."""
    for agent_name, agent_config in config.agents.items():
        if _uses_default_mind_workspace_scaffold(agent_name, agent_config):
            _ensure_default_mind_workspace(storage_path)


def agent_build_can_overlap_file_memory(agent_name: str, config: Config, storage_path: Path) -> bool:
    """Return True when it is safe to overlap agent construction with file-memory reads."""
    agent_config = config.get_agent(agent_name)
    if not _uses_default_mind_workspace_scaffold(agent_name, agent_config):
        return True
    memory_path = agent_workspace_root_path(storage_path, agent_name) / "MEMORY.md"
    return memory_path.is_file()


def _get_datetime_context(
    timezone_str: str,
    *,
    datetime_context_template: str,
) -> str:
    """Generate current date context for the agent.

    Args:
        timezone_str: Timezone string (e.g., 'America/New_York', 'UTC')
        datetime_context_template: Prompt template used for the rendered date context.

    Returns:
        Formatted string with current date and timezone information

    """
    tz = ZoneInfo(timezone_str)
    now = datetime.now(tz)

    date_str = now.strftime("%A, %B %d, %Y")
    timezone_abbrev = now.tzname() or timezone_str

    return render_prompt_template(
        datetime_context_template,
        date_str=date_str,
        timezone_str=timezone_str,
        timezone_abbrev=timezone_abbrev,
    )


def _get_mind_runtime_context(agent_name: str, runtime_paths: constants.RuntimePaths) -> str:
    """Render live installation facts for the default Mind agent."""
    if agent_name != _DEFAULT_MIND_AGENT_NAME:
        return ""
    homeserver = constants.runtime_matrix_homeserver(runtime_paths=runtime_paths)
    return f"\n## MindRoom Runtime\n- Active Matrix homeserver: `{homeserver}`\n"


def _load_context_files(
    context_files: list[Path | str],
    runtime_paths: constants.RuntimePaths,
    agent_name: str | None = None,
    storage_path: Path | None = None,
) -> list[_AdditionalContextChunk]:
    """Load configured context files."""
    loaded_parts: list[_AdditionalContextChunk] = []
    for raw_path in context_files:
        if isinstance(raw_path, Path):
            resolved_path = raw_path
        elif raw_path.startswith(_PROJECTED_WORKER_ASSET_PATH_PREFIXES):
            resolved_path = constants.resolve_config_relative_path(raw_path, runtime_paths)
        elif agent_name is not None and storage_path is not None:
            resolved_path = resolve_agent_owned_path(
                raw_path,
                agent_name=agent_name,
                base_storage_path=storage_path,
            )
        else:
            resolved_path = constants.resolve_config_relative_path(raw_path, runtime_paths)
        if resolved_path.is_file():
            body = _read_context_file(resolved_path)
            loaded_parts.append(
                # The title is the full path so the rendered prompt tells the
                # model exactly which file on disk each part came from.
                _AdditionalContextChunk(
                    title=str(resolved_path),
                    body=body,
                ),
            )
        else:
            logger.warning("context_file_not_found", agent=agent_name, path=str(resolved_path))
    return loaded_parts


@timed("system_prompt_assembly.agent_create.context_file_read")
def _read_context_file(resolved_path: Path) -> str:
    return resolved_path.read_text(encoding="utf-8").strip()


def _render_context_chunk(chunk: _AdditionalContextChunk, *, chunk_marker_template: str) -> str:
    """Render one chunk under its source path, marking what the cap removed."""
    body = chunk.body.strip()
    if not chunk.omitted_chars:
        return f"### {chunk.title}\n{body}" if body else ""
    marker = render_prompt_template(
        chunk_marker_template,
        title=chunk.title,
        omitted_chars=chunk.omitted_chars,
    )
    return f"### {chunk.title}\n{body}\n{marker}" if body else f"### {chunk.title}\n{marker}"


def _render_context_chunks(
    chunks: list[_AdditionalContextChunk],
    *,
    section_heading: str,
    chunk_marker_template: str,
) -> str:
    """Render every chunk that still carries content or an omission marker."""
    rendered = [
        block
        for chunk in chunks
        if (block := _render_context_chunk(chunk, chunk_marker_template=chunk_marker_template))
    ]
    if not rendered:
        return ""
    return f"{section_heading}\n" + "\n\n".join(rendered) + "\n\n"


def _drop_whole_chunks(
    chunks: list[_AdditionalContextChunk],
    max_preload_chars: int,
    *,
    render: Callable[[list[_AdditionalContextChunk]], str],
) -> int:
    """Drop entire chunk bodies (least critical first) until under the cap."""
    omitted = 0
    remaining_body_chunks = sum(bool(chunk.body) for chunk in chunks)
    for chunk in chunks:
        if len(render(chunks)) <= max_preload_chars:
            return omitted
        if not chunk.body:
            continue
        if remaining_body_chunks == 1:
            return omitted
        omitted += len(chunk.body)
        chunk.omitted_chars += len(chunk.body)
        chunk.body = ""
        remaining_body_chunks -= 1
    return omitted


def _trim_chunk_tails(
    chunks: list[_AdditionalContextChunk],
    max_preload_chars: int,
    *,
    render: Callable[[list[_AdditionalContextChunk]], str],
) -> int:
    """Trim from the *end* of chunks to preserve headers/identity at the top."""
    omitted = 0
    for chunk in chunks:
        while chunk.body:
            overflow = len(render(chunks)) - max_preload_chars
            if overflow <= 0:
                return omitted
            remove_count = min(overflow, len(chunk.body))
            original_length = len(chunk.body)
            chunk.body = chunk.body[: original_length - remove_count].rstrip()
            actual_removed = original_length - len(chunk.body)
            chunk.omitted_chars += actual_removed
            omitted += actual_removed
    return omitted


def _apply_preload_cap(
    personality_chunks: list[_AdditionalContextChunk],
    max_preload_chars: int,
    *,
    section_heading: str,
    truncation_marker_template: str,
    chunk_marker_template: str,
) -> tuple[str, int]:
    """Apply hard preload cap with deterministic truncation priority.

    Truncation order is by file list order.
    First drops whole chunks, then trims from the *end* of remaining chunks.
    Every touched file keeps a marker naming it and the chars it lost, so a
    dropped context file is visible to the model instead of silently missing.
    """
    render = partial(
        _render_context_chunks,
        section_heading=section_heading,
        chunk_marker_template=chunk_marker_template,
    )
    rendered = render(personality_chunks)
    if len(rendered) <= max_preload_chars:
        return rendered, 0

    # Trim against a budget that already reserves room for the summary marker,
    # so file bodies cannot consume space required by omission metadata.
    marker_upper_bound = render_prompt_template(
        truncation_marker_template,
        omitted_chars=sum(len(chunk.body) for chunk in personality_chunks),
    )
    marker_upper_bound_block = f"\n\n{marker_upper_bound}\n\n"
    required_marker_chunks = [
        replace(
            chunk,
            body="",
            omitted_chars=chunk.omitted_chars + len(chunk.body),
        )
        for chunk in personality_chunks
    ]
    minimum_rendered = render(required_marker_chunks).rstrip("\n") + marker_upper_bound_block
    if len(minimum_rendered) > max_preload_chars:
        msg = (
            f"max_preload_chars={max_preload_chars} cannot fit required context headings and omission markers; "
            f"at least {len(minimum_rendered)} characters are required"
        )
        raise ValueError(msg)
    trim_budget = max_preload_chars - len(marker_upper_bound_block)

    omitted_chars = _drop_whole_chunks(
        personality_chunks,
        trim_budget,
        render=render,
    )
    omitted_chars += _trim_chunk_tails(
        personality_chunks,
        trim_budget,
        render=render,
    )

    rendered = render(personality_chunks)
    if omitted_chars <= 0:
        return rendered, 0

    marker = render_prompt_template(truncation_marker_template, omitted_chars=omitted_chars)
    marker_block = f"\n\n{marker}\n\n"
    return rendered.rstrip("\n") + marker_block, omitted_chars


@timed("system_prompt_assembly.agent_create.additional_context")
def _build_additional_context(
    agent_name: str,
    agent_config: AgentConfig,
    max_preload_chars: int,
    *,
    personality_section_heading: str,
    truncation_marker_template: str,
    chunk_marker_template: str,
    workspace_context_files: tuple[Path, ...] = (),
    storage_path: Path,
    runtime_paths: constants.RuntimePaths,
) -> str:
    """Build additional role context from configured files/directories.

    This is evaluated each time one agent instance is created.
    The normal Matrix and OpenAI-compatible request paths build fresh agent
    instances per reply/request, so edits in the canonical agent workspace are
    reflected on the next reply without a process restart.
    """
    personality_chunks: list[_AdditionalContextChunk] = []
    context_files: list[Path | str] = [*agent_config.context_files, *workspace_context_files]
    if context_files:
        personality_chunks = _load_context_files(
            context_files,
            runtime_paths,
            agent_name,
            storage_path,
        )

    additional_context, omitted_chars = _apply_preload_cap(
        personality_chunks,
        max_preload_chars,
        section_heading=personality_section_heading,
        truncation_marker_template=truncation_marker_template,
        chunk_marker_template=chunk_marker_template,
    )
    if omitted_chars > 0:
        logger.warning(
            "Preload context exceeded max_preload_chars and was truncated",
            omitted_chars=omitted_chars,
            max_preload_chars=max_preload_chars,
        )
    return additional_context


def _file_mode_knowledge_instruction_block(
    agent_name: str,
    config: Config,
    agent_runtime: ResolvedAgentRuntime,
) -> str | None:
    """Describe workspace-accessible file-only knowledge bases to the model."""
    workspace = agent_runtime.workspace
    if workspace is None:
        return None

    lines = [
        "File-only knowledge bases are available in the workspace.",
        "Use file or coding tools such as grep, list, and read instead of search_knowledge_base for these sources.",
        "Available file-only sources:",
    ]
    source_lines: list[str] = []
    for base_id in config.resolve_entity(agent_name).knowledge_base_ids:
        base_config = config.get_knowledge_base_config(base_id)
        if base_config.mode != "files":
            continue
        relative_path = Path("knowledge") / base_id
        resolved_path = workspace.root / relative_path
        if not resolved_path.exists():
            continue
        description = " ".join(base_config.description.split()) or "No description configured."
        source_lines.append(f"- {base_id}: `{relative_path.as_posix()}`. {description}")
    if not source_lines:
        return None
    return "\n".join([*lines, *source_lines])


def _tool_supports_base_dir(tool_name: str) -> bool:
    """Return whether a registered tool exposes a base_dir config field."""
    metadata = TOOL_METADATA.get(tool_name)
    if metadata is None or not metadata.config_fields:
        return False
    return any(field.name == "base_dir" for field in metadata.config_fields)


def _tool_base_dir_override(
    tool_name: str,
    *,
    workspace_path: Path | None,
) -> dict[str, object] | None:
    """Build per-agent tool overrides for workspace-aware local tools."""
    if workspace_path is None or not _tool_supports_base_dir(tool_name):
        return None
    return {"base_dir": str(workspace_path)}


def _build_registered_agent_tool(
    tool_name: str,
    runtime_paths: constants.RuntimePaths,
    credentials_manager: CredentialsManager,
    shared_storage_path: Path,
    worker_tools: list[str],
    worker_scope: WorkerScope | None,
    allowed_shared_services: frozenset[str] | None,
    agent_name: str,
    tool_config_overrides: dict[str, object] | None,
    workspace_path: Path | None,
    tool_output_auto_save_threshold_bytes: int,
    routing_agent_is_private: bool,
    execution_identity: ToolExecutionIdentity | None,
    runtime_overrides: dict[str, object] | None,
) -> Toolkit:
    """Build one registered toolkit using the resolved routing inputs for this agent."""
    worker_target = build_agent_toolkit_worker_target(
        worker_scope,
        agent_name,
        is_private=routing_agent_is_private,
        execution_identity=execution_identity,
        runtime_paths=runtime_paths,
    )

    return get_tool_by_name(
        tool_name,
        runtime_paths,
        credentials_manager=credentials_manager,
        tool_config_overrides=tool_config_overrides,
        tool_init_overrides=_tool_base_dir_override(
            tool_name,
            workspace_path=workspace_path,
        ),
        runtime_overrides=runtime_overrides,
        shared_storage_root_path=shared_storage_path,
        worker_tools_override=worker_tools,
        allowed_shared_services=allowed_shared_services,
        tool_output_workspace_root=workspace_path,
        tool_output_auto_save_threshold_bytes=tool_output_auto_save_threshold_bytes,
        worker_target=worker_target,
    )


def _log_toolkits_without_unique_model_functions(
    toolkits: list[Toolkit],
    *,
    agent_name: str,
) -> None:
    """Warn when Agno would drop every function from a configured toolkit."""
    for parse_mode in ("sync", "async"):
        seen_function_names: set[str] = set()
        for toolkit in toolkits:
            functions = toolkit.get_async_functions() if parse_mode == "async" else toolkit.get_functions()
            function_names = set(functions)
            if function_names and function_names <= seen_function_names:
                logger.warning(
                    "Configured toolkit exposes no unique model functions because function names collide",
                    agent=agent_name,
                    toolkit=toolkit.name,
                    parse_mode=parse_mode,
                    function_names=sorted(function_names),
                )
            seen_function_names.update(function_names)


def _agent_tool_output_file_policy(
    agent_runtime: ResolvedAgentRuntime,
    runtime_paths: constants.RuntimePaths,
    tool_output_auto_save_threshold_bytes: int,
) -> ToolOutputFilePolicy | None:
    """Resolve the shared output-file policy for one agent's in-process tools."""
    if agent_runtime.tool_base_dir is None:
        return None
    return ToolOutputFilePolicy.from_runtime(
        agent_runtime.tool_base_dir,
        runtime_paths,
        auto_save_threshold_bytes=tool_output_auto_save_threshold_bytes,
    )


def _wrap_direct_agent_toolkit_for_output_files(
    toolkit: Toolkit,
    *,
    agent_runtime: ResolvedAgentRuntime,
    runtime_paths: constants.RuntimePaths,
    tool_output_auto_save_threshold_bytes: int,
) -> Toolkit:
    """Apply the central output-file wrapper to MindRoom-owned direct toolkits."""
    policy = _agent_tool_output_file_policy(
        agent_runtime,
        runtime_paths,
        tool_output_auto_save_threshold_bytes,
    )
    return wrap_toolkit_for_output_files(toolkit, policy)


@timed("system_prompt_assembly.agent_create.model_instance")
def _load_agent_model_instance(
    config: Config,
    runtime_paths: constants.RuntimePaths,
    model_name: str,
    execution_identity: ToolExecutionIdentity | None = None,
) -> Model:
    """Load one agent model while preserving prompt-assembly timing attribution."""
    return model_loading.get_model_instance(
        config,
        runtime_paths,
        model_name,
        execution_identity=execution_identity,
    )


@timed("system_prompt_assembly.agent_create.toolkit_build")
def build_agent_toolkit(  # noqa: C901, PLR0911, PLR0912
    tool_name: str,
    *,
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    worker_tools: list[str],
    runtime_overrides: dict[str, object] | None,
    agent_runtime: ResolvedAgentRuntime | None = None,
    tool_config_overrides: dict[str, object] | None = None,
    execution_identity: ToolExecutionIdentity | None,
    session_id: str | None = None,
    delegation_depth: int = 0,
    refresh_scheduler: KnowledgeRefreshScheduler | None = None,
    dynamic_tool_continuation: bool = False,
) -> Toolkit | None:
    """Build one configured toolkit for an agent.

    Callers own runtime override resolution before invoking this builder.
    Returns ``None`` when the configured tool should be skipped, such as an
    explicit ``delegate`` entry without valid delegation targets.

    ``dynamic_tool_continuation`` is only set by callers that resume the turn
    after a load/unload (the standalone agent envelope and materialized team
    members). It stops the dynamic tools manager's provider loop so the rebuilt
    agent sees the new schema.
    """
    agent_config = config.get_agent(agent_name)
    if agent_runtime is None:
        agent_runtime = resolve_agent_runtime(
            agent_name,
            config,
            runtime_paths,
            execution_identity=execution_identity,
            create=True,
        )
    storage_path = runtime_paths.storage_root
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    shared_storage_path = shared_storage_root(storage_path)

    if tool_name == "memory":
        if config.resolve_entity(agent_name).memory_backend == "none":
            return None

        from mindroom.custom_tools.memory import MemoryTools  # noqa: PLC0415

        # MemoryTools resolves the canonical per-agent storage roots internally via the
        # shared memory facade, so it should receive the caller-visible runtime root here.
        return _wrap_direct_agent_toolkit_for_output_files(
            MemoryTools(
                agent_name=agent_name,
                storage_path=storage_path,
                config=config,
                runtime_paths=runtime_paths,
                execution_identity=execution_identity,
            ),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "delegate":
        # Imported lazily to avoid a circular import through DelegateTools -> create_agent.
        from mindroom.custom_tools import delegate  # noqa: PLC0415

        if not agent_config.delegate_to:
            logger.warning(
                "Skipping delegate tool because delegate_to is empty",
                agent=agent_name,
            )
            return None
        if delegation_depth >= delegate.MAX_DELEGATION_DEPTH:
            logger.warning(
                "Skipping delegate tool because delegation depth limit was reached",
                agent=agent_name,
                delegation_depth=delegation_depth,
                max_delegation_depth=delegate.MAX_DELEGATION_DEPTH,
            )
            return None
        return _wrap_direct_agent_toolkit_for_output_files(
            delegate.DelegateTools(
                agent_name=agent_name,
                delegate_to=agent_config.delegate_to,
                runtime_paths=runtime_paths,
                config=config,
                execution_identity=execution_identity,
                delegation_depth=delegation_depth,
                refresh_scheduler=refresh_scheduler,
            ),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "self_config":
        from mindroom.custom_tools.self_config import SelfConfigTools  # noqa: PLC0415

        return _wrap_direct_agent_toolkit_for_output_files(
            SelfConfigTools(agent_name=agent_name, runtime_paths=runtime_paths),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "compact_context":
        from mindroom.custom_tools.compact_context import CompactContextTools  # noqa: PLC0415

        return _wrap_direct_agent_toolkit_for_output_files(
            CompactContextTools(
                agent_name=agent_name,
                config=config,
                runtime_paths=runtime_paths,
                execution_identity=execution_identity,
            ),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "dynamic_workflow":
        from mindroom.custom_tools.dynamic_workflow import DynamicWorkflowTools  # noqa: PLC0415

        return _wrap_direct_agent_toolkit_for_output_files(
            DynamicWorkflowTools(),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "report_publishing":
        from mindroom.custom_tools.report_publishing import ReportPublishingTools  # noqa: PLC0415

        return _wrap_direct_agent_toolkit_for_output_files(
            ReportPublishingTools(),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    if tool_name == "dynamic_tools":
        from mindroom.custom_tools.dynamic_tools import DynamicToolsToolkit  # noqa: PLC0415

        if not has_deferred_tools(config, agent_name):
            logger.warning(
                "Skipping 'dynamic_tools' tool for agent '%s': no deferred tools are configured",
                agent_name,
            )
            return None
        hidden_tool_names = _context_hidden_toolkits(execution_identity)
        if not _visible_deferred_tool_names(config, agent_name, hidden_tool_names=hidden_tool_names):
            logger.warning(
                "Skipping 'dynamic_tools' tool for agent '%s': no compatible deferred tools are available",
                agent_name,
            )
            return None
        if session_id is None:
            logger.warning(
                "Skipping 'dynamic_tools' tool for agent '%s': no stable session_id is available",
                agent_name,
            )
            return None
        return _wrap_direct_agent_toolkit_for_output_files(
            DynamicToolsToolkit(
                agent_name=agent_name,
                config=config,
                session_id=session_id,
                stop_after_tool_call=dynamic_tool_continuation,
                hidden_tool_names=hidden_tool_names,
            ),
            agent_runtime=agent_runtime,
            runtime_paths=runtime_paths,
            tool_output_auto_save_threshold_bytes=config.defaults.tool_output_auto_save_threshold_bytes,
        )

    return _build_registered_agent_tool(
        tool_name,
        runtime_paths,
        credentials_manager,
        shared_storage_path,
        worker_tools,
        agent_runtime.execution.execution_scope,
        (config.get_worker_grantable_credentials() if agent_runtime.execution.execution_scope is not None else None),
        agent_name,
        tool_config_overrides,
        agent_runtime.tool_base_dir,
        config.defaults.tool_output_auto_save_threshold_bytes,
        agent_runtime.execution.is_private,
        execution_identity,
        runtime_overrides,
    )


def get_agent_toolkit_names(
    agent_name: str,
    config: Config,
    *,
    session_id: str | None = None,
    delegation_depth: int = 0,
) -> list[str]:
    """Return the provider-visible toolkit list for a new agent runtime."""
    return [
        entry.name
        for entry in visible_tool_surface(
            agent_name=agent_name,
            config=config,
            session_id=session_id,
            delegation_depth=delegation_depth,
            enable_dynamic_tools_manager=session_id is not None,
        ).runtime_tool_configs
    ]


def resolve_runtime_worker_tools(
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    runtime_tool_names: list[str],
    *,
    tool_registry_preloaded: bool = False,
) -> list[str]:
    """Return worker-routed tools for one concrete runtime tool selection."""
    agent_config = config.get_agent(agent_name)
    configured = agent_config.worker_tools
    if configured is None:
        configured = config.defaults.worker_tools
    if configured is not None:
        return config.expand_tool_names(list(configured))

    if not tool_registry_preloaded:
        ensure_tool_registry_loaded(runtime_paths, config)
    return default_worker_routed_tools(runtime_tool_names)


def _render_tool_execution_environment(
    *,
    runtime_paths: constants.RuntimePaths,
    local_tool_names: tuple[str, ...],
    worker_routed_tool_names: tuple[str, ...],
    worker_scope: WorkerScope | None,
) -> str:
    """Describe effective per-tool execution routing to the model."""

    def tool_list(names: tuple[str, ...]) -> str:
        return ", ".join(f"`{name}`" for name in names) if names else "none"

    if not worker_routed_tool_names:
        if not local_tool_names:
            return "## Tool Execution Environment\n- No tools are available in this runtime."
        return (
            "## Tool Execution Environment\n"
            f"- All available tools run in the primary MindRoom runtime: {tool_list(local_tool_names)}.\n"
            "- No tools use a worker runtime."
        )

    backend = primary_worker_backend_name(runtime_paths)
    lines = [
        "## Tool Execution Environment",
        f"- Local tools (primary MindRoom runtime): {tool_list(local_tool_names)}.",
        f"- Worker-routed tools: {tool_list(worker_routed_tool_names)}.",
        f"- Worker backend: `{backend}`.",
    ]
    if backend == "static_runner":
        lines.append(
            "- Worker reuse: requests use the configured static runner; no per-user or per-agent "
            "worker state boundary is selected.",
        )
    else:
        scope_description = {
            None: "one runtime for this agent within the current tenant or account",
            "shared": "one runtime for this agent, shared by all users",
            "user": "one runtime for this user, shared across this user's agents",
            "user_agent": "one runtime used only by this user-agent pair",
        }[worker_scope]
        idle_behavior = {
            "docker": "the container stops",
            "kubernetes": "the deployment scales to zero",
        }[backend]
        lines.extend(
            (
                f"- Worker reuse: {scope_description}.",
                f"- Worker state is reused across turns for that scope. After the configured idle timeout, "
                f"{idle_behavior}; persisted files and caches remain until an operator deletes that worker state.",
            ),
        )
    lines.append("- Execution location is determined per tool; this agent is not sandboxed as a whole.")
    return "\n".join(lines)


def _registry_tool_routes_through_worker(
    tool_name: str,
    *,
    runtime_paths: constants.RuntimePaths,
    worker_tools: list[str],
) -> bool:
    """Return whether one successfully built registry toolkit uses worker routing."""
    metadata = TOOL_METADATA.get(tool_name)
    return (
        metadata is not None
        and metadata.factory is not None
        and sandbox_proxy_enabled_for_tool(
            tool_name,
            runtime_paths=runtime_paths,
            worker_tools_override=worker_tools,
        )
    )


def _is_learning_enabled(agent_config: AgentConfig, defaults: DefaultsConfig) -> bool:
    """Check if learning is enabled for an agent, falling back to defaults."""
    learning = agent_config.learning if agent_config.learning is not None else defaults.learning
    return learning is not False


def _context_hidden_toolkits(execution_identity: ToolExecutionIdentity | None) -> frozenset[str]:
    if execution_identity is None or execution_identity.room_id is not None:
        return frozenset()
    return frozenset(tool_name for tool_name, metadata in TOOL_METADATA.items() if metadata.requires_room_context)


def _visible_deferred_tool_names(
    config: Config,
    agent_name: str,
    *,
    hidden_tool_names: frozenset[str] = frozenset(),
) -> list[str]:
    return [
        entry.name
        for entry in config.resolve_entity(agent_name).authored_deferred_tool_configs
        if entry.name not in hidden_tool_names
    ]


def _resolve_agent_learning(
    agent_config: AgentConfig,
    defaults: DefaultsConfig,
    learning_storage: BaseDb | None = None,
) -> bool | LearningMachine:
    """Resolve Agent.learning setting from MindRoom agent configuration."""
    if not _is_learning_enabled(agent_config, defaults):
        return False

    learning_mode = agent_config.learning_mode or defaults.learning_mode
    learning_mode_value = LearningMode.AGENTIC if learning_mode == "agentic" else LearningMode.ALWAYS

    return LearningMachine(
        db=learning_storage,
        user_profile=UserProfileConfig(mode=learning_mode_value),
        user_memory=UserMemoryConfig(mode=learning_mode_value),
    )


def _build_dynamic_tooling_instruction_block(
    config: Config,
    agent_name: str,
    *,
    enable_dynamic_tools_manager: bool,
    hidden_tool_names: frozenset[str] = frozenset(),
) -> str | None:
    """Return static prompt guidance for per-tool dynamic loading."""
    if not enable_dynamic_tools_manager or not has_deferred_tools(config, agent_name):
        return None

    catalog_lines: list[str] = []
    deferred_entries = [
        entry
        for entry in deferred_tool_catalog_entries(agent_name=agent_name, config=config, loaded_tools=[])
        if entry.name not in hidden_tool_names
    ]
    if not deferred_entries:
        return None

    for entry in deferred_entries:
        description = entry.description.strip()
        if description:
            catalog_lines.append(f"{entry.name} - {description}")
        else:
            catalog_lines.append(entry.name)

    return config.render_prompt(
        "DYNAMIC_TOOLING_INSTRUCTION_TEMPLATE",
        tool_catalog="\n".join(catalog_lines),
    )


def _build_native_tool_search_instruction_blocks(
    config: Config,
    deferred_tool_names: tuple[str, ...],
) -> tuple[str, ...]:
    """Return compact discovery guidance for provider-native deferred tools."""
    if not deferred_tool_names:
        return ()
    return (
        config.render_prompt(
            "NATIVE_TOOL_SEARCH_INSTRUCTION_TEMPLATE",
            tool_domains=", ".join(deferred_tool_names),
        ),
    )


def _build_dynamic_tooling_state_suffix(
    config: Config,
    agent_name: str,
    *,
    loaded_tools: tuple[str, ...],
    enable_dynamic_tools_manager: bool,
    hidden_tool_names: frozenset[str] = frozenset(),
) -> str | None:
    """Return the small volatile dynamic-tool state suffix for the final instruction slot."""
    if not enable_dynamic_tools_manager or not has_deferred_tools(config, agent_name):
        return None
    if not _visible_deferred_tool_names(config, agent_name, hidden_tool_names=hidden_tool_names):
        return None

    initial_tools = [
        entry.name
        for entry in config.resolve_entity(agent_name).authored_deferred_tool_configs
        if entry.initial and entry.name not in hidden_tool_names
    ]
    visible_loaded_tools = tuple(tool_name for tool_name in loaded_tools if tool_name not in hidden_tool_names)
    current_tools = ", ".join(visible_loaded_tools) if visible_loaded_tools else "(none)"
    sticky_tools = ", ".join(initial_tools) if initial_tools else "(none)"
    return "\n".join(
        (
            "Dynamic tools currently loaded for this session: " + current_tools,
            "Sticky initial dynamic tools that cannot be unloaded: " + sticky_tools,
        ),
    )


def enable_all_history_replay(entity: Agent | Team) -> None:
    """Undo Agno's default three-run history fallback."""
    entity.num_history_runs = None


def remove_run_by_event_id(
    storage: BaseDb,
    session_id: str,
    event_id: str,
    *,
    session_type: SessionType = SessionType.AGENT,
    include_seen_event_ids: bool = False,
    remove_following_runs: bool = False,
) -> bool:
    """Remove a run whose Matrix source identity or consumed history matches.

    Redaction cleanup can also remove the causal suffix after the first match,
    because later model output may depend on content from the matching run.
    Returns True if any run was removed.
    """
    session = (
        agent_storage.get_team_session(storage, session_id)
        if session_type is SessionType.TEAM
        else agent_storage.get_agent_session(
            storage,
            session_id,
        )
    )
    if session is None or not session.runs:
        return False
    original_len = len(session.runs)
    filtered_runs: list[Any] = []
    matched_run = False
    for run in session.runs:
        if matched_run and remove_following_runs:
            continue
        if not isinstance(run, (RunOutput, TeamRunOutput)) or not run.metadata:
            filtered_runs.append(run)
            continue
        raw_source_event_ids = run.metadata.get(constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY)
        raw_discovery_event_ids = run.metadata.get(constants.MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY)
        raw_seen_event_ids = run.metadata.get(constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY)
        source_event_ids = (
            [candidate for candidate in raw_source_event_ids if isinstance(candidate, str)]
            if isinstance(raw_source_event_ids, list)
            else []
        )
        discovery_event_ids = (
            [candidate for candidate in raw_discovery_event_ids if isinstance(candidate, str)]
            if isinstance(raw_discovery_event_ids, list)
            else []
        )
        seen_event_ids = (
            [candidate for candidate in raw_seen_event_ids if isinstance(candidate, str)]
            if include_seen_event_ids and isinstance(raw_seen_event_ids, list)
            else []
        )
        matches_event_id = run.metadata.get(constants.MATRIX_EVENT_ID_METADATA_KEY) == event_id
        if (
            matches_event_id
            or event_id in source_event_ids
            or event_id in discovery_event_ids
            or event_id in seen_event_ids
        ):
            matched_run = True
            continue
        filtered_runs.append(run)
    session.runs = filtered_runs
    if len(session.runs) == original_len:
        return False
    storage.upsert_session(session)
    return True


def _resolve_culture_settings(mode: CultureMode) -> _CultureAgentSettings:
    """Map a culture mode to Agno culture feature flags."""
    if mode == "automatic":
        return _CultureAgentSettings(
            add_culture_to_context=True,
            update_cultural_knowledge=True,
            enable_agentic_culture=False,
        )
    if mode == "agentic":
        return _CultureAgentSettings(
            add_culture_to_context=True,
            update_cultural_knowledge=False,
            enable_agentic_culture=True,
        )
    return _CultureAgentSettings(
        add_culture_to_context=True,
        update_cultural_knowledge=False,
        enable_agentic_culture=False,
    )


def _culture_signature(culture_config: CultureConfig) -> tuple[str, str]:
    return (culture_config.mode, culture_config.description)


@timed("system_prompt_assembly.agent_create.culture_manager")
def _resolve_agent_culture(
    agent_name: str,
    config: Config,
    storage_path: Path,
    model: Model,
    *,
    cache_private: bool = False,
) -> tuple[CultureManager | None, _CultureAgentSettings | None]:
    """Resolve shared culture manager and feature flags for an agent."""
    culture_assignment = config.resolve_entity(agent_name).culture
    if culture_assignment is None:
        return None, None

    culture_name, culture_config = culture_assignment
    settings = _resolve_culture_settings(culture_config.mode)
    cache_key = (str(storage_path.resolve()), culture_name)
    signature = _culture_signature(culture_config)
    if cache_private:
        private_cache_key = (*cache_key, signature)
        cached_private_manager = _PRIVATE_CULTURE_MANAGER_CACHE.get(private_cache_key)
        if cached_private_manager is not None:
            cached_private_manager.model = model
            return cached_private_manager, settings
    else:
        cached_manager = _CULTURE_MANAGER_CACHE.get(cache_key)
        if cached_manager is not None and cached_manager.signature == signature:
            cached_manager.manager.model = model
            return cached_manager.manager, settings

    culture_scope = culture_config.description.strip() or "Shared best practices and principles."
    culture_manager = CultureManager(
        model=model,
        db=agent_storage.create_culture_storage(culture_name, storage_path),
        culture_capture_instructions=f"Culture '{culture_name}': {culture_scope}",
        add_knowledge=culture_config.mode != "manual",
        update_knowledge=culture_config.mode != "manual",
        delete_knowledge=False,
        clear_knowledge=False,
    )
    if cache_private:
        _PRIVATE_CULTURE_MANAGER_CACHE[private_cache_key] = culture_manager
    else:
        _CULTURE_MANAGER_CACHE[cache_key] = _CachedCultureManager(
            signature=signature,
            manager=culture_manager,
        )
    return culture_manager, settings


@timed("system_prompt_assembly.agent_create.load_plugins")
def _load_agent_plugins(config: Config, runtime_paths: constants.RuntimePaths) -> list[HookRegistryPlugin]:
    return cast("list[HookRegistryPlugin]", load_plugins(config, runtime_paths))


@timed("system_prompt_assembly.agent_create.tool_registry_sync")
def _sync_agent_tool_registry(config: Config, runtime_paths: constants.RuntimePaths) -> None:
    ensure_tool_registry_loaded(runtime_paths, config, load_plugin_tools=False)


@timed("system_prompt_assembly.agent_create.hook_bridge")
def _build_agent_tool_hook_bridge(
    *,
    hook_registry: HookRegistry | None,
    plugins: list[HookRegistryPlugin],
    agent_name: str,
    dispatch_context: ToolDispatchContext | None,
    config: Config,
    runtime_paths: constants.RuntimePaths,
) -> Callable[..., Any] | None:
    active_hook_registry = hook_registry if hook_registry is not None else HookRegistry.from_plugins(plugins)
    return build_tool_hook_bridge(
        active_hook_registry,
        agent_name=agent_name,
        dispatch_context=dispatch_context,
        config=config,
        runtime_paths=runtime_paths,
    )


def _prune_openai_incompatible_tools(
    toolkit: Toolkit,
    *,
    config: Config,
    execution_identity: ToolExecutionIdentity | None,
) -> Toolkit | None:
    """Hide tools from OpenAI-compatible agents when `/v1` cannot run them."""
    if execution_identity is None or execution_identity.channel != "openai_compat":
        return toolkit

    hidden_tool_names = {
        tool_name
        for tool_name in (*toolkit.functions, *toolkit.async_functions)
        if tool_requires_approval_for_openai_compat(config, tool_name)
    }
    if not hidden_tool_names:
        return toolkit

    toolkit.functions = {
        tool_name: function for tool_name, function in toolkit.functions.items() if tool_name not in hidden_tool_names
    }
    toolkit.async_functions = {
        tool_name: function
        for tool_name, function in toolkit.async_functions.items()
        if tool_name not in hidden_tool_names
    }

    if toolkit.functions or toolkit.async_functions:
        return toolkit
    return None


def _prune_toolkit_functions(
    toolkit: Toolkit,
    tool_function_filter: Callable[[Function], bool] | None,
) -> Toolkit | None:
    """Apply the final function policy and discard toolkits with no callable surface."""
    if tool_function_filter is not None:
        toolkit.functions = {
            name: function for name, function in toolkit.functions.items() if tool_function_filter(function)
        }
        toolkit.async_functions = {
            name: function for name, function in toolkit.async_functions.items() if tool_function_filter(function)
        }
    return toolkit if toolkit.functions or toolkit.async_functions else None


@timed("system_prompt_assembly.agent_create.dynamic_tool_selection")
def _resolve_agent_dynamic_tool_selection(
    *,
    agent_name: str,
    config: Config,
    session_id: str | None,
    delegation_depth: int,
    native_deferred_tools: bool,
    eager_deferred_tools: bool,
) -> VisibleToolSurface:
    if native_deferred_tools or eager_deferred_tools:
        # Attach every authored deferred tool and skip the dynamic-tools
        # manager when the consuming runtime cannot rebuild its tool schema.
        return visible_tool_surface(
            agent_name=agent_name,
            config=config,
            loaded_tools=_visible_deferred_tool_names(config, agent_name),
            delegation_depth=delegation_depth,
            enable_dynamic_tools_manager=False,
        )
    return resolve_dynamic_tool_selection(
        agent_name=agent_name,
        config=config,
        session_id=session_id,
        delegation_depth=delegation_depth,
    )


def _render_agent_identity_context(
    agent_name: str,
    display_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    *,
    model_provider: str,
    model_id: str,
    include_openai_compat_guidance: bool,
) -> str:
    openai_compat_history_guidance = config.get_prompt("OPENAI_COMPAT_HISTORY_GUIDANCE")
    if include_openai_compat_guidance:
        openai_context = render_prompt_template(
            config.get_prompt("OPENAI_COMPAT_AGENT_IDENTITY_CONTEXT_TEMPLATE"),
            agent_name=agent_name,
            display_name=display_name,
            model_provider=model_provider,
            model_id=model_id,
            openai_compat_history_guidance=openai_compat_history_guidance,
        )
        has_custom_matrix_identity = "AGENT_IDENTITY_CONTEXT_TEMPLATE" in config.prompts
        has_custom_openai_identity = "OPENAI_COMPAT_AGENT_IDENTITY_CONTEXT_TEMPLATE" in config.prompts
        if not has_custom_matrix_identity or has_custom_openai_identity:
            return openai_context
        overridden_identity_context = render_prompt_template(
            config.get_prompt("AGENT_IDENTITY_CONTEXT_TEMPLATE"),
            display_name=display_name,
            matrix_id="not available in OpenAI-compatible API",
            model_provider=model_provider,
            model_id=model_id,
            openai_compat_history_guidance=openai_compat_history_guidance,
        )
        return overridden_identity_context + openai_context

    matrix_id = entity_identity_registry(config, runtime_paths).current_id(agent_name).full_id
    return build_agent_identity_context(
        display_name=display_name,
        matrix_id=matrix_id,
        model_provider=model_provider,
        model_id=model_id,
        include_openai_compat_guidance=False,
        identity_context_template=config.get_prompt("AGENT_IDENTITY_CONTEXT_TEMPLATE"),
        openai_compat_history_guidance=openai_compat_history_guidance,
    )


@timed("system_prompt_assembly.agent_create.skills_load")
def _load_agent_skills(
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    *,
    workspace_skills_root: Path | None = None,
    output_file_policy: ToolOutputFilePolicy | None = None,
) -> Skills | None:
    return build_agent_skills(
        agent_name,
        config,
        runtime_paths,
        workspace_skills_root=workspace_skills_root,
        output_file_policy=output_file_policy,
    )


@timed("system_prompt_assembly.agent_create.agent_init")
def _initialize_agent_instance(**agent_kwargs: Any) -> Agent:  # noqa: ANN401
    knowledge_sources = cast(
        "tuple[KnowledgeSourceDescription, ...]",
        agent_kwargs.pop("knowledge_sources", ()),
    )
    tool_function_filter = cast(
        "Callable[[Function], bool] | None",
        agent_kwargs.pop("tool_function_filter", None),
    )
    agent = Agent(**agent_kwargs)
    agent.knowledge_sources = knowledge_sources
    agent.tool_function_filter = tool_function_filter
    return agent


def _agent_create_timing(label: str, **event_data: object) -> AbstractContextManager[None]:
    return timed_block(f"system_prompt_assembly.agent_create.{label}", scope=None, **event_data)


def _assemble_agent_toolkits(
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    agent_runtime: ResolvedAgentRuntime,
    *,
    execution_identity: ToolExecutionIdentity | None,
    session_id: str | None,
    hook_registry: HookRegistry | None,
    disable_runtime_capabilities: bool,
    disabled_tool_names: frozenset[str],
    tool_function_filter: Callable[[Function], bool] | None,
    delegation_depth: int,
    refresh_scheduler: KnowledgeRefreshScheduler | None,
    dynamic_tool_continuation: bool,
    native_deferred_tools: bool,
    eager_deferred_tools: bool,
) -> _AgentToolAssembly:
    """Assemble runtime toolkits and the dynamic-tool visibility for one agent instance."""
    plugins = _load_agent_plugins(config, runtime_paths)
    _sync_agent_tool_registry(config, runtime_paths)
    tool_hook_bridge = _build_agent_tool_hook_bridge(
        hook_registry=hook_registry,
        plugins=plugins,
        agent_name=agent_name,
        dispatch_context=(
            ToolDispatchContext(execution_identity=execution_identity) if execution_identity is not None else None
        ),
        config=config,
        runtime_paths=runtime_paths,
    )
    # Dynamic tool state is keyed by agent and session scope, so team members
    # sharing one Matrix thread do not leak loaded tools across agents.
    dynamic_tool_selection = _resolve_agent_dynamic_tool_selection(
        agent_name=agent_name,
        config=config,
        session_id=session_id,
        delegation_depth=delegation_depth,
        native_deferred_tools=native_deferred_tools,
        eager_deferred_tools=eager_deferred_tools,
    )
    hidden_toolkits = _context_hidden_toolkits(execution_identity)
    resolved_tool_configs = {entry.name: entry for entry in dynamic_tool_selection.runtime_tool_configs}
    if disable_runtime_capabilities:
        resolved_tool_configs = {}
    elif disabled_tool_names:
        resolved_tool_configs = {
            tool_name: entry
            for tool_name, entry in resolved_tool_configs.items()
            if tool_name not in disabled_tool_names
        }
    if hidden_toolkits:
        resolved_tool_configs = {
            tool_name: entry for tool_name, entry in resolved_tool_configs.items() if tool_name not in hidden_toolkits
        }
    loaded_tools = (
        ()
        if disable_runtime_capabilities
        else tuple(
            tool_name
            for tool_name in dynamic_tool_selection.loaded_tools
            if tool_name not in disabled_tool_names and tool_name not in hidden_toolkits
        )
    )
    with _agent_create_timing("resolve_worker_tools"):
        worker_tools = resolve_runtime_worker_tools(
            agent_name,
            config,
            runtime_paths,
            list(resolved_tool_configs),
            tool_registry_preloaded=True,
        )
    entity_view = config.resolve_entity(agent_name)
    tools: list[Toolkit] = []
    local_tool_names: list[str] = []
    worker_routed_tool_names: list[str] = []
    deferred_tool_names: list[str] = []
    deferred_wire_tool_names: set[str] = set()
    for tool_name, tool_entry in resolved_tool_configs.items():
        try:
            runtime_overrides = entity_view.tool_runtime_overrides(tool_name)
            with _agent_create_timing("toolkit_build.one", tool_name=tool_name):
                toolkit = build_agent_toolkit(
                    tool_name,
                    agent_name=agent_name,
                    config=config,
                    runtime_paths=runtime_paths,
                    worker_tools=worker_tools,
                    runtime_overrides=runtime_overrides,
                    agent_runtime=agent_runtime,
                    tool_config_overrides=tool_entry.tool_config_overrides,
                    session_id=session_id,
                    execution_identity=execution_identity,
                    delegation_depth=delegation_depth,
                    refresh_scheduler=refresh_scheduler,
                    dynamic_tool_continuation=dynamic_tool_continuation,
                )
            if toolkit:
                toolkit = _prune_openai_incompatible_tools(
                    toolkit,
                    config=config,
                    execution_identity=execution_identity,
                )
            if toolkit:
                toolkit = _prune_toolkit_functions(toolkit, tool_function_filter)
            if toolkit:
                toolkit = prepend_tool_hook_bridge(toolkit, tool_hook_bridge)
                tools.append(toolkit)
                target_names = (
                    worker_routed_tool_names
                    if _registry_tool_routes_through_worker(
                        tool_name,
                        runtime_paths=runtime_paths,
                        worker_tools=worker_tools,
                    )
                    else local_tool_names
                )
                target_names.append(tool_name)
                # initial deferred tools were always loaded, so they stay in
                # the rendered prompt prefix as plain non-deferred tools.
                if native_deferred_tools and tool_entry.defer and not tool_entry.initial:
                    suppress_fully_deferred_toolkit_instructions(toolkit)
                    deferred_tool_names.append(tool_entry.authored_name or tool_name)
                    deferred_wire_tool_names.update(toolkit.get_functions())
                    deferred_wire_tool_names.update(toolkit.get_async_functions())
        except (ValueError, ImportError) as exc:
            logger.warning(
                "Could not load tool for agent construction",
                tool=tool_name,
                agent=agent_name,
                error=str(exc),
            )
    return _AgentToolAssembly(
        tools=tools,
        loaded_tools=loaded_tools,
        hidden_toolkits=hidden_toolkits,
        selected_dynamic_tools=dynamic_tool_selection.loaded_tools,
        deferred_tool_names=tuple(dict.fromkeys(deferred_tool_names)),
        deferred_wire_tool_names=frozenset(deferred_wire_tool_names),
        local_tool_names=tuple(local_tool_names),
        worker_routed_tool_names=tuple(worker_routed_tool_names),
    )


def _open_agent_session_storage(
    agent_name: str,
    agent_runtime: ResolvedAgentRuntime,
    *,
    history_storage: BaseDb | None,
    persist_runtime_state: bool,
) -> BaseDb | None:
    """Open (or reuse) durable Agno session storage for one agent instance."""
    with _agent_create_timing("storage_open"):
        if not persist_runtime_state:
            return None
        if history_storage is not None:
            return history_storage
        return agent_storage.create_state_storage(
            agent_name,
            agent_runtime.state_root,
            subdir="sessions",
            session_table=f"{agent_name}_sessions",
        )


def _build_agent_role_context(
    agent_name: str,
    agent_config: AgentConfig,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    agent_runtime: ResolvedAgentRuntime,
    *,
    active_model_name: str | None,
    include_openai_compat_guidance: bool,
    disable_runtime_capabilities: bool,
    local_tool_names: tuple[str, ...],
    worker_routed_tool_names: tuple[str, ...],
) -> _AgentRoleContext:
    """Resolve the model name and render identity, datetime, and preload context into the role."""
    # Get model config for identity context
    model_name = active_model_name or agent_config.model or "default"
    if model_name in config.models:
        model_config = config.models[model_name]
        model_provider = model_config.provider.title()  # Capitalize provider name
        model_id = model_config.id
    else:
        # Fallback if model not found
        model_provider = "AI"
        model_id = model_name

    with _agent_create_timing("identity_context"):
        identity_context = _render_agent_identity_context(
            agent_name,
            agent_config.display_name,
            config,
            runtime_paths,
            model_provider=model_provider,
            model_id=model_id,
            include_openai_compat_guidance=include_openai_compat_guidance,
        )

    # Add current date context with the user's configured timezone
    datetime_context = _get_datetime_context(
        config.timezone,
        datetime_context_template=config.get_prompt("DATETIME_CONTEXT_TEMPLATE"),
    )

    # Combine identity, datetime, and live installation contexts.
    full_context = identity_context + datetime_context + _get_mind_runtime_context(agent_name, runtime_paths)

    if not disable_runtime_capabilities:
        full_context += "\n\n" + _render_tool_execution_environment(
            runtime_paths=runtime_paths,
            local_tool_names=local_tool_names,
            worker_routed_tool_names=worker_routed_tool_names,
            worker_scope=agent_runtime.execution.execution_scope,
        )
        workspace = agent_runtime.workspace
        full_context += _build_additional_context(
            agent_name,
            agent_config,
            config.defaults.max_preload_chars,
            personality_section_heading=config.get_prompt("PERSONALITY_CONTEXT_SECTION_HEADING"),
            truncation_marker_template=config.get_prompt("CONTEXT_TRUNCATION_MARKER_TEMPLATE"),
            chunk_marker_template=config.get_prompt("CONTEXT_CHUNK_OMITTED_MARKER_TEMPLATE"),
            workspace_context_files=workspace.context_files if workspace is not None else (),
            storage_path=runtime_paths.storage_root,
            runtime_paths=runtime_paths,
        )

    return _AgentRoleContext(model_name=model_name, role=full_context + agent_config.role)


def _build_agent_instructions(
    agent_name: str,
    agent_config: AgentConfig,
    config: Config,
    agent_runtime: ResolvedAgentRuntime,
    *,
    skills: Skills | None,
    session_id: str | None,
    include_interactive_questions: bool,
    disable_runtime_capabilities: bool,
    hidden_toolkits: frozenset[str],
    loaded_tools: tuple[str, ...],
    native_deferred_tool_names: tuple[str, ...],
    all_deferred_tools_eager: bool,
) -> list[str]:
    """Accumulate the configured and runtime instruction blocks for one agent instance."""
    instructions = list(agent_config.instructions)

    if skills and skills.get_skill_names():
        instructions.append(config.get_prompt("SKILLS_TOOL_USAGE_PROMPT"))

    instructions.extend(
        _build_native_tool_search_instruction_blocks(
            config,
            native_deferred_tool_names,
        ),
    )

    # Native server-side tool search replaces the load_tool catalog and
    # loaded-state prompt blocks, so the native path emits neither.
    enable_dynamic_tools_manager = session_id is not None and not all_deferred_tools_eager
    dynamic_tooling_block = None
    if not disable_runtime_capabilities:
        dynamic_tooling_block = _build_dynamic_tooling_instruction_block(
            config,
            agent_name,
            enable_dynamic_tools_manager=enable_dynamic_tools_manager,
            hidden_tool_names=hidden_toolkits,
        )
    if dynamic_tooling_block is not None:
        instructions.append(dynamic_tooling_block)

    if agent_runtime.tool_base_dir is not None and not disable_runtime_capabilities:
        instructions.append(config.get_prompt("OUTPUT_REDIRECT_PROMPT"))
        instructions.append(config.get_prompt("WORKSPACE_SKILL_AUTHORING_PROMPT"))

    file_mode_knowledge_instruction = (
        None
        if disable_runtime_capabilities
        else _file_mode_knowledge_instruction_block(agent_name, config, agent_runtime)
    )
    if file_mode_knowledge_instruction is not None:
        instructions.append(file_mode_knowledge_instruction)

    show_tool_calls = show_tool_calls_for_agent(config, agent_name)
    if not show_tool_calls:
        instructions.append(config.get_prompt("HIDDEN_TOOL_CALLS_PROMPT"))

    if include_interactive_questions:
        instructions.append(config.get_prompt("INTERACTIVE_QUESTION_PROMPT"))

    dynamic_tooling_state_suffix = None
    if not disable_runtime_capabilities:
        dynamic_tooling_state_suffix = _build_dynamic_tooling_state_suffix(
            config,
            agent_name,
            loaded_tools=loaded_tools,
            enable_dynamic_tools_manager=enable_dynamic_tools_manager,
            hidden_tool_names=hidden_toolkits,
        )
    if dynamic_tooling_state_suffix is not None:
        instructions.append(dynamic_tooling_state_suffix)

    return instructions


def _resolve_agent_culture_state(
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    agent_runtime: ResolvedAgentRuntime,
    model: Model,
    *,
    persist_runtime_state: bool,
) -> _AgentCultureState:
    """Resolve the culture manager and Agent culture flags for one agent instance."""
    if not persist_runtime_state:
        return _AgentCultureState(
            manager=None,
            add_culture_to_context=None,
            update_cultural_knowledge=False,
            enable_agentic_culture=False,
        )

    culture_storage_root = runtime_paths.storage_root
    cache_private_culture = False
    if agent_runtime.execution.is_private:
        worker_key = agent_runtime.execution.worker_key
        if worker_key is None:
            msg = f"Private agent '{agent_name}' requires a worker key to resolve culture state"
            raise ValueError(msg)
        execution_scope = agent_runtime.execution.execution_scope
        execution_identity = agent_runtime.execution.execution_identity
        if execution_scope is None or execution_identity is None:
            msg = f"Private agent '{agent_name}' requires an execution scope and identity to resolve culture state"
            raise ValueError(msg)
        culture_storage_root = resolve_private_requester_scope_root(
            runtime_paths=runtime_paths,
            execution_scope=execution_scope,
            execution_identity=execution_identity,
            worker_key=worker_key,
        )
        cache_private_culture = True

    culture_manager, culture_settings = _resolve_agent_culture(
        agent_name,
        config,
        culture_storage_root,
        model,
        cache_private=cache_private_culture,
    )
    add_culture_to_context: bool | None = None
    update_cultural_knowledge = False
    enable_agentic_culture = False
    if culture_settings is not None:
        add_culture_to_context = culture_settings.add_culture_to_context
        update_cultural_knowledge = culture_settings.update_cultural_knowledge
        enable_agentic_culture = culture_settings.enable_agentic_culture
    return _AgentCultureState(
        manager=culture_manager,
        add_culture_to_context=add_culture_to_context,
        update_cultural_knowledge=update_cultural_knowledge,
        enable_agentic_culture=enable_agentic_culture,
    )


@timed("system_prompt_assembly.agent_create")
def create_agent(
    agent_name: str,
    config: Config,
    runtime_paths: constants.RuntimePaths,
    execution_identity: ToolExecutionIdentity | None,
    *,
    session_id: str | None = None,
    hook_registry: HookRegistry | None = None,
    knowledge: KnowledgeProtocol | None = None,
    history_storage: BaseDb | None = None,
    active_model_name: str | None = None,
    include_interactive_questions: bool = True,
    include_openai_compat_guidance: bool = False,
    persist_runtime_state: bool = True,
    disable_runtime_capabilities: bool = False,
    disabled_tool_names: frozenset[str] = frozenset(),
    tool_function_filter: Callable[[Function], bool] | None = None,
    delegation_depth: int = 0,
    refresh_scheduler: KnowledgeRefreshScheduler | None = None,
    dynamic_tool_continuation: bool = False,
    eager_deferred_tools: bool = False,
) -> Agent:
    """Create an agent instance from configuration.

    Args:
        agent_name: Name of the agent to create
        config: Application configuration
        runtime_paths: Explicit runtime context for paths, env, and credentials.
        execution_identity: Request execution identity used to resolve scoped
            state, workspaces, worker routing, and requester-local storage.
        session_id: Stable Agno session id used to resolve session-scoped
            dynamic tool state.
        hook_registry: Optional hook registry for plugin-based tool call
            interception and event hooks.
        knowledge: Optional shared knowledge base instance for RAG-enabled agents.
        history_storage: Optional already-open session storage to reuse for this agent.
        active_model_name: Optional runtime-selected model name overriding the configured model.
        include_interactive_questions: Whether to include the interactive
            question authoring prompt. Set to False for channels that do not
            support Matrix reaction-based question flows.
        include_openai_compat_guidance: Whether to include OpenAI-compatible
            history-format guidance in the shared identity prompt.
        persist_runtime_state: Whether this agent instance should write durable
            Agno history, learning, and culture state.
        disable_runtime_capabilities: Whether to omit tools, skills, knowledge,
            and preloaded context files for a restricted in-process agent run.
        disabled_tool_names: Resolved tool names to omit from this instance.
        tool_function_filter: Optional policy applied to each fully resolved
            function after normal toolkit construction.
        delegation_depth: Current delegation nesting depth. Used to prevent
            infinite recursion when agents delegate to each other.
        refresh_scheduler: Optional runtime-owned shared knowledge refresh scheduler
            passed through to delegated child agents.
        dynamic_tool_continuation: Whether this agent runs under a response-turn
            loop that resumes after a dynamic load/unload (the standalone agent
            envelope and materialized team members both do). Embedded agents
            without such a loop leave it False so a load/unload takes effect on
            the next request instead of truncating the run.
        eager_deferred_tools: Whether to materialize every deferred toolkit and
            omit the dynamic-tools manager for a runtime with an immutable tool
            schema.

    Returns:
        Configured Agent instance

    Raises:
        ValueError: If agent_name is not found in configuration

    """
    create_runtime_state = not disable_runtime_capabilities
    with _agent_create_timing("resolve_runtime"):
        agent_runtime = resolve_agent_runtime(
            agent_name,
            config,
            runtime_paths,
            execution_identity=execution_identity,
            create=create_runtime_state,
        )

    agent_config = config.get_agent(agent_name)
    if create_runtime_state:
        ensure_default_agent_workspaces(config, runtime_paths.storage_root)
    defaults = config.defaults

    # Gate on this agent's resolved runtime model (thread overrides and team
    # members resolve per agent), not on any surrounding team's model.
    runtime_model_config = config.models.get(active_model_name or agent_config.model or "default")
    native_deferred_tools = runtime_model_config is not None and (
        native_tool_search_supported(runtime_model_config.provider, runtime_model_config.id)
        or openai_native_tool_search_supported(
            runtime_model_config.provider,
            runtime_model_config.id,
            base_url=(runtime_model_config.extra_kwargs or {}).get("base_url")
            or runtime_paths.env_value("OPENAI_BASE_URL"),
        )
    )

    tool_assembly = _assemble_agent_toolkits(
        agent_name,
        config,
        runtime_paths,
        agent_runtime,
        execution_identity=execution_identity,
        session_id=session_id,
        hook_registry=hook_registry,
        disable_runtime_capabilities=disable_runtime_capabilities,
        disabled_tool_names=disabled_tool_names,
        tool_function_filter=tool_function_filter,
        delegation_depth=delegation_depth,
        refresh_scheduler=refresh_scheduler,
        dynamic_tool_continuation=dynamic_tool_continuation,
        native_deferred_tools=native_deferred_tools,
        eager_deferred_tools=eager_deferred_tools,
    )
    storage = _open_agent_session_storage(
        agent_name,
        agent_runtime,
        history_storage=history_storage,
        persist_runtime_state=persist_runtime_state,
    )
    learning_storage = (
        agent_storage.create_state_storage(
            storage_name=agent_name,
            state_root=agent_runtime.state_root,
            subdir="learning",
            session_table=f"{agent_name}_learning_sessions",
        )
        if persist_runtime_state and _is_learning_enabled(agent_config, defaults)
        else None
    )

    role_context = _build_agent_role_context(
        agent_name,
        agent_config,
        config,
        runtime_paths,
        agent_runtime,
        active_model_name=active_model_name,
        include_openai_compat_guidance=include_openai_compat_guidance,
        disable_runtime_capabilities=disable_runtime_capabilities,
        local_tool_names=tool_assembly.local_tool_names,
        worker_routed_tool_names=tool_assembly.worker_routed_tool_names,
    )

    # Create agent with defaults applied
    model = _load_agent_model_instance(config, runtime_paths, role_context.model_name, execution_identity)
    if tool_assembly.deferred_wire_tool_names:
        # Each installer no-ops on the other provider family's model class.
        install_claude_deferred_tool_search(model, deferred_tool_names=tool_assembly.deferred_wire_tool_names)
        install_openai_deferred_tool_search(model, deferred_tool_names=tool_assembly.deferred_wire_tool_names)
    logger.info(
        "create_agent",
        agent=agent_name,
        model_class=model.__class__.__name__,
        model_id=model.id,
        native_deferred_tools=sorted(tool_assembly.deferred_wire_tool_names),
    )

    workspace = agent_runtime.workspace
    skills = (
        None
        if disable_runtime_capabilities
        else _load_agent_skills(
            agent_name,
            config,
            runtime_paths,
            workspace_skills_root=workspace.root / "skills" if workspace is not None else None,
            output_file_policy=_agent_tool_output_file_policy(
                agent_runtime,
                runtime_paths,
                config.defaults.tool_output_auto_save_threshold_bytes,
            ),
        )
    )
    instructions = _build_agent_instructions(
        agent_name,
        agent_config,
        config,
        agent_runtime,
        skills=skills,
        session_id=session_id,
        include_interactive_questions=include_interactive_questions,
        disable_runtime_capabilities=disable_runtime_capabilities,
        hidden_toolkits=tool_assembly.hidden_toolkits,
        loaded_tools=tool_assembly.loaded_tools,
        native_deferred_tool_names=tool_assembly.deferred_tool_names,
        all_deferred_tools_eager=native_deferred_tools or eager_deferred_tools,
    )

    _log_toolkits_without_unique_model_functions(tool_assembly.tools, agent_name=agent_name)

    entity_view = config.resolve_entity(agent_name)
    knowledge_enabled = not disable_runtime_capabilities and knowledge is not None
    knowledge_sources = (
        knowledge_source_descriptions(knowledge) if knowledge_enabled and isinstance(knowledge, Knowledge) else ()
    )
    culture = _resolve_agent_culture_state(
        agent_name,
        config,
        runtime_paths,
        agent_runtime,
        model,
        persist_runtime_state=persist_runtime_state,
    )

    # Shared history-policy source of truth with the team replay path.
    history_settings = entity_view.history_settings
    history_policy = history_settings.policy

    compress_tool_results = (
        agent_config.compress_tool_results
        if agent_config.compress_tool_results is not None
        else defaults.compress_tool_results
    )

    agent = _initialize_agent_instance(
        name=agent_config.display_name,
        id=agent_name,
        role=role_context.role,
        model=model,
        tools=tool_assembly.tools,
        skills=skills,
        instructions=instructions,
        db=storage,
        learning=_resolve_agent_learning(agent_config, defaults, learning_storage) if persist_runtime_state else False,
        markdown=agent_config.markdown if agent_config.markdown is not None else defaults.markdown,
        knowledge=knowledge if knowledge_enabled else None,
        knowledge_sources=knowledge_sources,
        tool_function_filter=tool_function_filter,
        search_knowledge=knowledge_enabled,
        add_history_to_context=persist_runtime_state,
        add_session_summary_to_context=persist_runtime_state,
        num_history_runs=history_policy.num_history_runs,
        num_history_messages=history_policy.num_history_messages,
        # Keep persisted runs raw even though Agno replays history natively.
        store_history_messages=False,
        culture_manager=culture.manager,
        add_culture_to_context=culture.add_culture_to_context,
        update_cultural_knowledge=culture.update_cultural_knowledge,
        enable_agentic_culture=culture.enable_agentic_culture,
        compress_tool_results=compress_tool_results,
        max_tool_calls_from_history=history_settings.max_tool_calls_from_history,
        telemetry=False,
    )
    if history_policy.mode == "all":
        enable_all_history_replay(agent)

    logger.info(
        "Created agent",
        agent=agent_name,
        display_name=agent_config.display_name,
        tool_count=len(tool_assembly.tools),
        loaded_dynamic_tools=list(tool_assembly.selected_dynamic_tools),
    )

    return agent


__all__ = [
    "agent_build_can_overlap_file_memory",
    "build_agent_toolkit",
    "create_agent",
    "describe_agent",
    "enable_all_history_replay",
    "ensure_default_agent_workspaces",
    "get_agent_toolkit_names",
    "remove_run_by_event_id",
    "resolve_runtime_worker_tools",
    "show_tool_calls_for_agent",
]
