dify/api/repositories/app_definition_query_repository.py

"""Database repository for externally visible app definitions."""

from typing import Any, cast, override

from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
from core.app.apps.agent_app.app_feature_projection import merge_agent_app_features
from core.app.apps.agent_app.app_variable_projection import agent_app_variables_to_user_input_form
from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
from models.account import Tenant, TenantStatus
from models.agent import AgentConfigSnapshot
from models.agent_config_entities import AgentSoulConfig
from models.model import App, AppMode, AppModelConfig, Site, load_annotation_reply_config
from models.tools import ApiToolProvider
from models.workflow import Workflow
from services.app_definition_query_service import (
    AppDefinitionQuery,
    AppDefinitionSummary,
    AppParameterConfig,
    AppSiteConfiguration,
    AppToolIconSource,
)
from services.web_app_runtime_query_service import WebAppRuntimeRecord


def _map_site_configuration(site: Site) -> AppSiteConfiguration:
    return AppSiteConfiguration(
        title=site.title,
        chat_color_theme=site.chat_color_theme,
        chat_color_theme_inverted=site.chat_color_theme_inverted,
        icon_type=site.icon_type.value if site.icon_type is not None else None,
        icon=site.icon,
        icon_background=site.icon_background,
        description=site.description,
        copyright=site.copyright,
        privacy_policy=site.privacy_policy,
        input_placeholder=site.input_placeholder,
        custom_disclaimer=site.custom_disclaimer,
        default_language=site.default_language,
        prompt_public=site.prompt_public,
        show_workflow_steps=site.show_workflow_steps,
        use_icon_as_answer_icon=site.use_icon_as_answer_icon,
    )


def _get_public_agent_parameter_config(app: App, *, session: Session) -> AppParameterConfig:
    app_model_config = app.app_model_config_with_session(session=session)
    agent = app.agent_app_binding_with_session(session=session)
    if agent is None:
        raise AgentAppGeneratorError("Agent App has no bound Agent")
    if not agent_has_workflow_callable_active_snapshot(session=session, agent=agent):
        raise AgentAppNotPublishedError("Agent has not been published")

    snapshot = session.scalar(
        select(AgentConfigSnapshot)
        .where(
            AgentConfigSnapshot.tenant_id == app.tenant_id,
            AgentConfigSnapshot.agent_id == agent.id,
            AgentConfigSnapshot.id == agent.active_config_snapshot_id,
        )
        .limit(1)
    )
    if snapshot is None:
        raise AgentAppGeneratorError("Agent published version not found")

    agent_soul = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
    annotation_reply = load_annotation_reply_config(session, app.id) if app_model_config else None
    return AppParameterConfig(
        features_dict=merge_agent_app_features(
            agent_soul=agent_soul,
            app_model_config=app_model_config,
            annotation_reply=annotation_reply,
        ),
        user_input_form=agent_app_variables_to_user_input_form(agent_soul.app_variables),
    )


class AppDefinitionQueryRepository(AppDefinitionQuery):
    def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
        self._session_factory = session_factory

    @override
    def get_published_parameter_config(
        self,
        app_id: str,
        *,
        public_runtime: bool = False,
    ) -> AppParameterConfig | None:
        with self._session_factory() as session:
            app = session.get(App, app_id)
            if app is None:
                return None

            if public_runtime and app.mode == AppMode.AGENT:
                return _get_public_agent_parameter_config(app, session=session)

            if app.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
                workflow = app.workflow_with_session(session=session)
                if workflow is None:
                    return None

                return AppParameterConfig(
                    features_dict=workflow.features_dict,
                    user_input_form=cast(list[dict[str, Any]], workflow.user_input_form(to_old_structure=True)),
                )

            app_model_config = app.app_model_config_with_session(session=session)
            if app_model_config is None:
                return None

            features_dict = app_model_config.to_dict(
                annotation_reply=load_annotation_reply_config(session, app.id),
            )
            return AppParameterConfig(
                features_dict=features_dict,
                user_input_form=cast(list[dict[str, Any]], features_dict.get("user_input_form", [])),
            )

    @override
    def get_tool_icon_sources(self, app_id: str) -> tuple[AppToolIconSource, ...] | None:
        with self._session_factory() as session:
            app = session.get(App, app_id)
            if app is None:
                return None

            records: list[AppToolIconSource] = []
            for tool in self._get_tools(session, app):
                if len(tool) < 4:
                    continue

                provider_type = str(tool.get("provider_type", ""))
                provider_id = str(tool.get("provider_id", ""))
                tool_name = str(tool.get("tool_name", ""))
                provider_icon: str | None = None
                if provider_type == "api":
                    try:
                        provider = session.get(ApiToolProvider, provider_id)
                        provider_icon = provider.icon if provider is not None else None
                    except Exception:
                        # Preserve the legacy response fallback when a provider cannot be loaded.
                        provider_icon = None

                records.append(
                    AppToolIconSource(
                        provider_type=provider_type,
                        provider_id=provider_id,
                        tool_name=tool_name,
                        provider_icon=provider_icon,
                    )
                )

            return tuple(records)

    @override
    def get_summary(self, app_id: str) -> AppDefinitionSummary | None:
        with self._session_factory() as session:
            app = session.get(App, app_id)
            if app is None:
                return None

            return AppDefinitionSummary(
                name=app.name,
                description=app.description,
                tags=tuple(tag.name for tag in app.tags_with_session(session=session)),
                mode=app.mode.value,
                author_name=app.author_name_with_session(session=session),
            )

    @override
    def get_site_configuration(self, app_id: str) -> AppSiteConfiguration | None:
        with self._session_factory() as session:
            site = session.scalar(select(Site).where(Site.app_id == app_id).limit(1))
            if site is None:
                return None

            return _map_site_configuration(site)

    def get_runtime_record(self, app_id: str) -> WebAppRuntimeRecord | None:
        with self._session_factory() as session:
            app = session.get(App, app_id)
            if app is None:
                return None

            site = session.scalar(select(Site).where(Site.app_id == app_id).limit(1))
            if site is None:
                return None

            tenant = session.get(Tenant, app.tenant_id)
            if tenant is None:
                return None

            app_id = app.id
            tenant_id = app.tenant_id
            enable_site = app.enable_site
            site_configuration = _map_site_configuration(site)
            plan = tenant.plan
            tenant_status = tenant.status.value
            tenant_custom_config_json = tenant.custom_config
            mode = AppMode.value_of(app.mode).value
            if tenant.status != TenantStatus.ARCHIVE:
                mode = AppMode.value_of(app.mode_compatible_with_agent_with_session(session=session)).value
            return WebAppRuntimeRecord(
                app_id=app_id,
                tenant_id=tenant_id,
                mode=mode,
                enable_site=enable_site,
                site=site_configuration,
                plan=plan,
                tenant_status=tenant_status,
                tenant_custom_config_json=tenant_custom_config_json,
            )

    @staticmethod
    def _get_tools(session: Session, app: App) -> list[dict[str, Any]]:
        if app.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
            workflow = session.get(Workflow, app.workflow_id) if app.workflow_id else None
            if workflow is None:
                return []

            tools: list[dict[str, Any]] = []
            nodes = cast(list[dict[str, Any]], workflow.graph_dict.get("nodes", []))
            for node in nodes:
                node_data = node.get("data", {})
                if node_data.get("type") == "tool":
                    tools.append(
                        {
                            "provider_type": node_data.get("provider_type"),
                            "provider_id": node_data.get("provider_id"),
                            "tool_name": node_data.get("tool_name"),
                            "tool_parameters": {},
                        }
                    )
            return tools

        app_model_config = session.get(AppModelConfig, app.app_model_config_id) if app.app_model_config_id else None
        if app_model_config is None:
            return []
        return cast(list[dict[str, Any]], app_model_config.agent_mode_dict.get("tools", []))