"""Build dify-agent run requests for one Agent App conversation turn.

Mirrors the workflow ``WorkflowAgentRuntimeRequestBuilder`` but for the Agent
App surface: the user prompt is the chat message (no workflow-node job / no
previous-node context), multi-turn continuity flows through the
conversation-keyed ``session_snapshot`` plus the history layer, and Agent Soul
knowledge config is mapped into the same fixed ``dify.knowledge_base`` layer
used by workflow runs.
"""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Literal, Protocol, cast

from agenton.compositor import CompositorSessionSnapshot
from dify_agent.layers.execution_context import (
    DifyExecutionContextInvokeFrom,
    DifyExecutionContextLayerConfig,
    DifyExecutionContextUserFrom,
)
from dify_agent.protocol import CreateRunRequest, DeferredToolResultsPayload

from clients.agent_backend import (
    AgentBackendAgentAppRunInput,
    AgentBackendModelConfig,
    AgentBackendRunRequestBuilder,
    redact_for_agent_backend_log,
)
from configs import dify_config
from core.app.entities.app_invoke_entities import DifyRunContext, InvokeFrom
from core.workflow.nodes.agent_v2.dify_tools_builder import (
    WorkflowAgentDifyToolLayersBuilder,
    WorkflowAgentDifyToolsBuilder,
    WorkflowAgentDifyToolsBuildError,
    WorkflowAgentToolLayers,
)
from core.workflow.nodes.agent_v2.runtime_request_builder import (
    append_runtime_warnings,
    build_ask_human_layer_config,
    build_config_aware_soul_mention_resolver,
    build_config_layer_config,
    build_knowledge_layer_config,
    build_shell_layer_config,
)
from models.agent_config_entities import AgentSoulConfig, AgentSoulToolsConfig
from models.provider_ids import ModelProviderID
from services.agent.prompt_mentions import expand_prompt_mentions


class AgentAppRuntimeRequestBuildError(ValueError):
    """Raised when Agent App state cannot be mapped to a valid run request."""

    def __init__(self, error_code: str, message: str) -> None:
        self.error_code = error_code
        super().__init__(message)


class CredentialsProvider(Protocol):
    def fetch(self, provider_name: str, model_name: str) -> dict[str, Any]: ...


@dataclass(frozen=True, slots=True)
class AgentAppRuntimeBuildContext:
    dify_context: DifyRunContext
    agent_id: str
    agent_config_snapshot_id: str
    agent_soul: AgentSoulConfig
    conversation_id: str
    user_query: str
    idempotency_key: str
    agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot"
    session_snapshot: CompositorSessionSnapshot | None = None
    # ENG-638: set when resuming a chat turn after a submitted ask_human form.
    deferred_tool_results: DeferredToolResultsPayload | None = None
    suspend_on_exit: bool = True


@dataclass(frozen=True, slots=True)
class AgentAppRuntimeRequest:
    request: CreateRunRequest
    redacted_request: dict[str, Any]
    metadata: dict[str, Any]


class AgentAppRuntimeRequestBuilder:
    """Build dify-agent run requests from Agent App conversation state."""

    def __init__(
        self,
        *,
        credentials_provider: CredentialsProvider,
        request_builder: AgentBackendRunRequestBuilder | None = None,
        dify_tools_builder: WorkflowAgentDifyToolLayersBuilder | None = None,
    ) -> None:
        self._credentials_provider = credentials_provider
        self._request_builder = request_builder or AgentBackendRunRequestBuilder()
        self._dify_tools_builder = dify_tools_builder or WorkflowAgentDifyToolsBuilder()

    def build(self, context: AgentAppRuntimeBuildContext) -> AgentAppRuntimeRequest:
        agent_soul = context.agent_soul
        if agent_soul.model is None:
            raise AgentAppRuntimeRequestBuildError(
                "agent_model_not_configured",
                "Agent App requires the Agent Soul model to be configured.",
            )

        metadata = self._build_metadata(context)
        credentials = self._credentials_provider.fetch(agent_soul.model.model_provider, agent_soul.model.model)
        try:
            tool_layers = self._build_tool_layers(
                tenant_id=context.dify_context.tenant_id,
                app_id=context.dify_context.app_id,
                user_id=context.dify_context.user_id,
                tools=agent_soul.tools,
                invoke_from=context.dify_context.invoke_from,
            )
        except WorkflowAgentDifyToolsBuildError as error:
            raise AgentAppRuntimeRequestBuildError(error.error_code, str(error)) from error
        if tool_layers.plugin_tools is not None or tool_layers.core_tools is not None or agent_soul.tools.cli_tools:
            metadata["agent_tools"] = {
                "dify_tool_count": len(tool_layers.exposed_tool_names()),
                "dify_tool_names": tool_layers.exposed_tool_names(),
                "cli_tool_count": len(agent_soul.tools.cli_tools),
            }

        config_layer_config, config_warnings = build_config_layer_config(
            agent_soul,
            agent_id=context.agent_id,
            config_version_id=context.agent_config_snapshot_id,
            config_version_kind=context.agent_config_version_kind,
        )
        append_runtime_warnings(metadata, config_warnings)
        soul_prompt_resolver = build_config_aware_soul_mention_resolver(agent_soul)
        knowledge_config = build_knowledge_layer_config(agent_soul)

        request = self._request_builder.build_for_agent_app(
            AgentBackendAgentAppRunInput(
                model=AgentBackendModelConfig(
                    plugin_id=self._plugin_daemon_plugin_id(
                        plugin_id=agent_soul.model.plugin_id,
                        model_provider=agent_soul.model.model_provider,
                    ),
                    model_provider=self._plugin_daemon_provider_name(agent_soul.model.model_provider),
                    model=agent_soul.model.model,
                    credentials=self._normalize_credentials(credentials),
                    model_settings=agent_soul.model.model_settings.model_dump(mode="json", exclude_none=True),
                ),
                execution_context=DifyExecutionContextLayerConfig(
                    tenant_id=context.dify_context.tenant_id,
                    user_id=context.dify_context.user_id,
                    app_id=context.dify_context.app_id,
                    conversation_id=context.conversation_id,
                    agent_id=context.agent_id,
                    agent_config_version_id=context.agent_config_snapshot_id,
                    agent_config_version_kind=context.agent_config_version_kind,
                    # Agent Files §1.3: real Dify access context + agent run mode.
                    user_from=cast(DifyExecutionContextUserFrom, context.dify_context.user_from.value),
                    invoke_from=cast(DifyExecutionContextInvokeFrom, context.dify_context.invoke_from.value),
                    agent_mode="agent_app",
                ),
                # ENG-616: expand slash-menu mention tokens to canonical names so
                # no frontend-internal {{#…#}} marker ever reaches the model.
                agent_soul_prompt=expand_prompt_mentions(agent_soul.prompt.system_prompt, soul_prompt_resolver).strip()
                or None,
                agent_config_version_kind=context.agent_config_version_kind,
                user_prompt=context.user_query,
                tools=tool_layers.plugin_tools,
                core_tools=tool_layers.core_tools,
                knowledge=knowledge_config,
                config_layer_config=config_layer_config,
                ask_human_config=build_ask_human_layer_config(agent_soul),
                include_shell=dify_config.AGENT_SHELL_ENABLED,
                shell_config=build_shell_layer_config(agent_soul),
                session_snapshot=context.session_snapshot,
                deferred_tool_results=context.deferred_tool_results,
                suspend_on_exit=context.suspend_on_exit,
                idempotency_key=context.idempotency_key,
                metadata=metadata,
            )
        )
        redacted = cast(dict[str, Any], redact_for_agent_backend_log(request))
        return AgentAppRuntimeRequest(request=request, redacted_request=redacted, metadata=metadata)

    def _build_tool_layers(
        self,
        *,
        tenant_id: str,
        app_id: str,
        user_id: str | None,
        tools: AgentSoulToolsConfig,
        invoke_from: InvokeFrom,
    ) -> WorkflowAgentToolLayers:
        # Production Agent App runs intentionally keep existing plugin configs
        # on the direct `dify.plugin.tools` route. This builder emits plugin
        # tools directly and non-plugin Dify tools through `dify.core.tools`.
        return self._dify_tools_builder.build_layers(
            tenant_id=tenant_id,
            app_id=app_id,
            user_id=user_id,
            tools=tools,
            invoke_from=invoke_from,
        )

    @staticmethod
    def _build_metadata(context: AgentAppRuntimeBuildContext) -> dict[str, Any]:
        return {
            "tenant_id": context.dify_context.tenant_id,
            "app_id": context.dify_context.app_id,
            "conversation_id": context.conversation_id,
            "agent_id": context.agent_id,
            "agent_config_snapshot_id": context.agent_config_snapshot_id,
        }

    @staticmethod
    def _plugin_daemon_plugin_id(*, plugin_id: str, model_provider: str) -> str:
        """Return the transport plugin id expected by plugin-daemon headers."""
        if plugin_id.count("/") == 1:
            return plugin_id.split(":", 1)[0].split("@", 1)[0]
        if plugin_id:
            return ModelProviderID(plugin_id).plugin_id
        return ModelProviderID(model_provider).plugin_id

    @staticmethod
    def _plugin_daemon_provider_name(model_provider: str) -> str:
        """Return the provider name expected by plugin-daemon dispatch payloads."""
        return ModelProviderID(model_provider).provider_name

    @staticmethod
    def _normalize_credentials(credentials: Mapping[str, Any]) -> dict[str, str | int | float | bool | None]:
        normalized: dict[str, str | int | float | bool | None] = {}
        for key, value in credentials.items():
            if isinstance(value, str | int | float | bool) or value is None:
                normalized[key] = value
            else:
                normalized[key] = str(value)
        return normalized


__all__ = [
    "AgentAppRuntimeBuildContext",
    "AgentAppRuntimeRequest",
    "AgentAppRuntimeRequestBuildError",
    "AgentAppRuntimeRequestBuilder",
]
