refactor(api): reuse app definition queries for app parameters (#40465)

2026-08-30 17:11:50 +08:00 · 2026-08-14 09:38:34 +00:00
parent 17c8ff6a33
commit e0fd3b264a
14 changed files with 345 additions and 435 deletions
@@ -1,60 +0,0 @@
-from typing import Any
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-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.agent import Agent, AgentConfigSnapshot, AgentStatus
-from models.agent_config_entities import AgentSoulConfig
-from models.model import App, load_annotation_reply_config
-
-
-def get_published_agent_app_feature_dict_and_user_input_form(
-    app_model: App,
-    *,
-    session: Session,
-) -> tuple[dict[str, Any], list[dict[str, Any]]]:
-    """Return public Agent App parameters backed by the published Agent Soul."""
-    app_model_config = app_model.app_model_config_with_session(session=session)
-
-    agent_id = app_model.bound_agent_id
-    if not agent_id:
-        raise AgentAppGeneratorError("Agent App has no bound Agent")
-
-    agent = session.scalar(
-        select(Agent)
-        .where(
-            Agent.tenant_id == app_model.tenant_id,
-            Agent.id == agent_id,
-            Agent.status == AgentStatus.ACTIVE,
-        )
-        .limit(1)
-    )
-    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_model.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_model.id) if app_model_config else None
-    features_dict = merge_agent_app_features(
-        agent_soul=agent_soul,
-        app_model_config=app_model_config,
-        annotation_reply=annotation_reply,
-    )
-    return features_dict, agent_app_variables_to_user_input_form(agent_soul.app_variables)
@@ -1,5 +1,4 @@
 import logging
-from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Literal
@@ -51,7 +50,6 @@ from controllers.console.files import FILE_UPLOAD_PARAMS, upload_file_from_reque
 from controllers.console.remote_files import RemoteFileUploadPayload, upload_remote_file_from_request
 from controllers.console.wraps import cloud_edition_billing_resource_check, model_validate, with_current_user
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import (
@@ -61,6 +59,7 @@ from core.errors.error import (
 )
 from core.helper import encrypter
 from core.workflow.llm_environment_variable import LLMEnvironmentVariable, dump_environment_variable
+from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.base import ResponseModel
@@ -74,9 +73,10 @@ from libs import helper
 from libs.helper import dump_response, to_timestamp, uuid_value
 from models import Account, App
 from models.account import TenantStatus
-from models.model import AppMode, Site, load_annotation_reply_config
+from models.model import AppMode, Site
 from models.workflow import Workflow
 from services.account_service import TenantService
+from services.app_definition_query_service import AppDefinitionUnavailableError
 from services.app_generate_service import AppGenerateService
 from services.app_ref_service import AppRefService
 from services.app_service import AppResponseView, AppService
@@ -855,26 +855,12 @@ class TrialAppParameterApi(Resource):
        if app_model is None:
            raise AppUnavailableError()

-        features_dict: Mapping[str, Any]
-        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow_with_session(session=session)
-            if workflow is None:
-                raise AppUnavailableError()
+        try:
+            parameters = application_services().app_definitions.get_parameters(app_model.id)
+        except AppDefinitionUnavailableError:
+            raise AppUnavailableError() from None

-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config_with_session(session=session)
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            annotation_reply = load_annotation_reply_config(session, app_model_config.app_id)
-            features_dict = app_model_config.to_dict(annotation_reply=annotation_reply)
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-        return ParametersResponse.model_validate(parameters).model_dump(mode="json")
+        return dump_response(ParametersResponse, parameters)


 class AppApi(Resource):
@@ -1,23 +1,18 @@
-from typing import Any, cast
+from typing import Any

 from flask_restx import Resource
 from pydantic import Field
-from sqlalchemy.orm import Session

-from controllers.common.agent_app_parameters import get_published_agent_app_feature_dict_and_user_input_form
 from controllers.common.fields import Parameters
 from controllers.common.schema import register_response_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import AgentNotPublishedError, AppUnavailableError
 from controllers.service_api.wraps import validate_app_token
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
 from extensions.ext_application_services import application_services
-from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import dump_response
-from models.model import App, AppMode, load_annotation_reply_config
-from services.app_definition_query_service import AppDefinitionUnavailableError
+from models.model import App
+from services.app_definition_query_service import AppDefinitionNotPublishedError, AppDefinitionUnavailableError


 class AppInfoResponse(ResponseModel):
@@ -35,19 +30,6 @@ class AppMetaResponse(ResponseModel):
 register_response_schema_models(service_api_ns, Parameters, AppMetaResponse, AppInfoResponse)


-def _get_agent_app_feature_dict_and_user_input_form(
-    app_model: App,
-    *,
-    session: Session,
-) -> tuple[dict[str, Any], list[dict[str, Any]]]:
-    try:
-        return get_published_agent_app_feature_dict_and_user_input_form(app_model, session=session)
-    except AgentAppNotPublishedError:
-        raise AgentNotPublishedError()
-    except AgentAppGeneratorError:
-        raise AppUnavailableError()
-
-
@service_api_ns.route("/parameters")
 class AppParameterApi(Resource):
    """Resource for app variables."""
@@ -80,36 +62,14 @@ class AppParameterApi(Resource):

        Returns the input form parameters and configuration for the application.
        """
-        session = db.session()
-        features_dict: dict[str, Any]
-        user_input_form: list[dict[str, Any]]
-        if app_model.mode == AppMode.AGENT:
-            features_dict, user_input_form = _get_agent_app_feature_dict_and_user_input_form(
-                app_model,
-                session=session,
-            )
-        elif app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow_with_session(session=session)
-            if workflow is None:
-                raise AppUnavailableError()
+        try:
+            parameters = application_services().app_definitions.get_public_parameters(app_model.id)
+        except AppDefinitionNotPublishedError:
+            raise AgentNotPublishedError() from None
+        except AppDefinitionUnavailableError:
+            raise AppUnavailableError() from None

-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config_with_session(session=session)
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            annotation_reply = load_annotation_reply_config(session, app_model.id)
-            features_dict = cast(
-                dict[str, Any],
-                app_model_config.to_dict(annotation_reply=annotation_reply),
-            )
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-        return Parameters.model_validate(parameters).model_dump(mode="json")
+        return dump_response(Parameters, parameters)


@service_api_ns.route("/meta")
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, cast
+from typing import Any

 from flask import request
 from flask_restx import Resource
@@ -8,17 +8,15 @@ from werkzeug.exceptions import Unauthorized

 from constants import HEADER_NAME_APP_CODE
 from controllers.common import fields
-from controllers.common.agent_app_parameters import get_published_agent_app_feature_dict_and_user_input_form
+from controllers.common.fields import Parameters
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
 from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from libs.helper import dump_response
 from libs.passport import PassportService
 from libs.token import extract_webapp_passport
-from models.model import App, AppMode, EndUser, load_annotation_reply_config
-from services.app_definition_query_service import AppDefinitionUnavailableError
+from models.model import App, EndUser
+from services.app_definition_query_service import AppDefinitionNotPublishedError, AppDefinitionUnavailableError
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
@@ -54,7 +52,7 @@ class AppMetaResponse(BaseModel):
 register_schema_models(web_ns, AppAccessModeQuery, AppPermissionQuery)
 register_response_schema_models(
    web_ns,
-    fields.Parameters,
+    Parameters,
    AppMetaResponse,
    fields.AccessModeResponse,
    fields.BooleanResultResponse,
@@ -77,44 +75,17 @@ class AppParameterApi(WebApiResource):
            500: "Internal Server Error",
        }
    )
-    @web_ns.response(200, "Success", web_ns.models[fields.Parameters.__name__])
+    @web_ns.response(200, "Success", web_ns.models[Parameters.__name__])
    def get(self, app_model: App, end_user: EndUser):
        """Retrieve app parameters."""
-        session = db.session()
-        features_dict: dict[str, Any]
-        user_input_form: list[dict[str, Any]]
-        if app_model.mode == AppMode.AGENT:
-            try:
-                features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
-                    app_model,
-                    session=session,
-                )
-            except AgentAppNotPublishedError:
-                raise AgentNotPublishedError()
-            except AgentAppGeneratorError:
-                raise AppUnavailableError()
-        elif app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow_with_session(session=session)
-            if workflow is None:
-                raise AppUnavailableError()
+        try:
+            parameters = application_services().app_definitions.get_public_parameters(app_model.id)
+        except AppDefinitionNotPublishedError:
+            raise AgentNotPublishedError() from None
+        except AppDefinitionUnavailableError:
+            raise AppUnavailableError() from None

-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config_with_session(session=session)
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            annotation_reply = load_annotation_reply_config(session, app_model.id)
-            features_dict = cast(
-                dict[str, Any],
-                app_model_config.to_dict(annotation_reply=annotation_reply),
-            )
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-        return fields.Parameters.model_validate(parameters).model_dump(mode="json")
+        return dump_response(Parameters, parameters)


@web_ns.route("/meta")
@@ -2,25 +2,72 @@

 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.agent import AgentConfigSnapshot
+from models.agent_config_entities import AgentSoulConfig
 from models.model import App, AppMode, AppModelConfig, load_annotation_reply_config
 from models.tools import ApiToolProvider
 from models.workflow import Workflow
 from services.app_definition_query_service import AppDefinitionQuery, AppParameterConfig, AppToolIconSource


+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) -> AppParameterConfig | None:
+    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:
@@ -5,6 +5,7 @@ from collections.abc import Mapping, Sequence
 from typing import Any, NamedTuple, Protocol

 from core.app.app_config.common.parameters_mapping import AppParametersDict, get_parameters_from_feature_dict
+from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError


 class AppParameterConfig(NamedTuple):
@@ -20,7 +21,12 @@ class AppToolIconSource(NamedTuple):


 class AppDefinitionQuery(Protocol):
-    def get_published_parameter_config(self, app_id: str) -> AppParameterConfig | None: ...
+    def get_published_parameter_config(
+        self,
+        app_id: str,
+        *,
+        public_runtime: bool = False,
+    ) -> AppParameterConfig | None: ...

    def get_tool_icon_sources(self, app_id: str) -> Sequence[AppToolIconSource] | None: ...

@@ -29,6 +35,10 @@ class AppDefinitionUnavailableError(ValueError):
    """Raised when an app definition is unavailable."""


+class AppDefinitionNotPublishedError(AppDefinitionUnavailableError):
+    """Raised when a public Agent App has not been published."""
+
+
 _API_TOOL_FALLBACK_ICON = {"background": "#252525", "content": "\ud83d\ude01"}


@@ -44,6 +54,21 @@ class AppDefinitionQueryService:

    def get_parameters(self, app_id: str) -> AppParametersDict:
        config = self._definitions.get_published_parameter_config(app_id)
+        return self._map_parameters(config)
+
+    def get_public_parameters(self, app_id: str) -> AppParametersDict:
+        """Read public parameters, using the published Soul for Agent Apps."""
+        try:
+            config = self._definitions.get_published_parameter_config(app_id, public_runtime=True)
+        except AgentAppNotPublishedError:
+            raise AppDefinitionNotPublishedError from None
+        except AgentAppGeneratorError:
+            raise AppDefinitionUnavailableError from None
+
+        return self._map_parameters(config)
+
+    @staticmethod
+    def _map_parameters(config: AppParameterConfig | None) -> AppParametersDict:
        if config is None:
            raise AppDefinitionUnavailableError

@@ -10,7 +10,7 @@ extend-select = ["ANN401", "ARG"]
 "commands/test_archive_workflow_runs.py" = ["ARG005"]
 "commands/test_data_migration_wizard.py" = ["ARG005"]
 "commands/test_legacy_model_type_migration.py" = ["ARG001", "ARG002", "ARG005"]
-"controllers/common/test_agent_app_parameters.py" = ["ARG005", "TID251"]
+"repositories/test_app_definition_query_repository_agent.py" = ["ARG005", "TID251"]
 "controllers/common/test_app_access.py" = ["ARG005"]
 "controllers/console/agent/test_agent_controllers.py" = ["ARG001", "ARG002", "ARG003", "ARG005", "TID251"]
 "controllers/console/app/test_agent_app_sandbox.py" = ["ARG002", "ARG005"]
@@ -30,6 +30,7 @@ from controllers.console.explore.error import (
 )
 from controllers.console.explore.trial import ChatRequest, CompletionRequest, TextToSpeechRequest, WorkflowRunRequest
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.errors.error import (
    ModelCurrentlyNotSupportError,
    ProviderTokenNotInitError,
@@ -109,23 +110,6 @@ def trial_app_workflow() -> MagicMock:
    return app


-@pytest.fixture
-def valid_parameters() -> dict[str, object]:
-    return {
-        "user_input_form": [],
-        "system_parameters": {},
-        "suggested_questions": {},
-        "suggested_questions_after_answer": {},
-        "speech_to_text": {},
-        "text_to_speech": {},
-        "retriever_resource": {},
-        "annotation_reply": {},
-        "more_like_this": {},
-        "sensitive_word_avoidance": {},
-        "file_upload": {},
-    }
-
-
 def test_trial_workflow_uses_trial_scoped_simple_account_model() -> None:
    assert module.simple_account_model.name == "TrialSimpleAccount"
    assert module.simple_account_model.__schema__["properties"].keys() >= {"id", "name", "email"}
@@ -888,63 +872,33 @@ class TestTrialAppParameterApi:
        with pytest.raises(AppUnavailableError):
            method(api, unbound_session, None)

-    def test_success_non_workflow(self, valid_parameters: dict[str, object], unbound_session: Session) -> None:
+    def test_success(self, unbound_session: Session) -> None:
        api = module.TrialAppParameterApi()
        method = unwrap(api.get)
+        parameters = get_parameters_from_feature_dict(features_dict={}, user_input_form=[])
+        expected = module.ParametersResponse.model_validate(parameters).model_dump(mode="json")
+        app_definitions = MagicMock()
+        app_definitions.get_parameters.return_value = parameters
+        services = SimpleNamespace(app_definitions=app_definitions)

-        app_model_config = MagicMock(app_id="app-1")
-        app_model_config.to_dict.return_value = {"user_input_form": []}
-        app_model = SimpleNamespace(
-            mode=AppMode.CHAT,
-            app_model_config_with_session=MagicMock(return_value=app_model_config),
-        )
-        annotation_reply = {"enabled": False}
+        with patch.object(module, "application_services", return_value=services):
+            result = method(api, unbound_session, SimpleNamespace(id="app-1"))

-        with (
-            patch.object(
-                module, "load_annotation_reply_config", return_value=annotation_reply
-            ) as load_annotation_reply,
-            patch.object(
-                module,
-                "get_parameters_from_feature_dict",
-                return_value=valid_parameters,
-            ),
-            patch.object(
-                module.ParametersResponse,
-                "model_validate",
-                return_value=MagicMock(model_dump=lambda mode=None: {"ok": True}),
-            ),
-        ):
-            result = method(api, unbound_session, app_model)
+        assert result == expected
+        app_definitions.get_parameters.assert_called_once_with("app-1")

-        assert result == {"ok": True}
-        app_model.app_model_config_with_session.assert_called_once_with(session=unbound_session)
-        load_annotation_reply.assert_called_once_with(unbound_session, "app-1")
-        app_model_config.to_dict.assert_called_once_with(annotation_reply=annotation_reply)
-
-    def test_success_workflow(self, valid_parameters: dict[str, object], unbound_session: Session) -> None:
+    def test_unavailable_parameters(self, unbound_session: Session) -> None:
        api = module.TrialAppParameterApi()
        method = unwrap(api.get)
+        app_definitions = MagicMock()
+        app_definitions.get_parameters.side_effect = module.AppDefinitionUnavailableError
+        services = SimpleNamespace(app_definitions=app_definitions)

-        workflow = MagicMock(features_dict={})
-        workflow.user_input_form.return_value = []
-        app_model = SimpleNamespace(
-            mode=AppMode.WORKFLOW,
-            workflow_with_session=MagicMock(return_value=workflow),
-        )
        with (
-            patch.object(module, "get_parameters_from_feature_dict", return_value=valid_parameters),
-            patch.object(
-                module.ParametersResponse,
-                "model_validate",
-                return_value=MagicMock(model_dump=lambda mode=None: {"ok": True}),
-            ),
+            patch.object(module, "application_services", return_value=services),
+            pytest.raises(AppUnavailableError),
        ):
-            result = method(api, unbound_session, app_model)
-
-        assert result == {"ok": True}
-        app_model.workflow_with_session.assert_called_once_with(session=unbound_session)
-        workflow.user_input_form.assert_called_once_with(to_old_structure=True)
+            method(api, unbound_session, SimpleNamespace(id="app-1"))


 class TestTrialChatAudioApi:
@@ -1,11 +1,10 @@
 """SQLite-backed tests for Service API application controllers.

 The authentication decorator resolves the app, tenant, and tenant owner before
-the controller runs.  Controller/model code then reads configuration, workflow,
-tags, and author information through two additional references to the database
-extension.  Tests bind all of those references to one explicit scoped SQLite
-session and persist visibility and cross-tenant decoys instead of fabricating ORM
-lookup results.
+the controller runs. Controller/model code then reads tags and author information
+through model database properties. Tests bind those references to one explicit
+scoped SQLite session and persist visibility and cross-tenant decoys instead of
+fabricating ORM lookup results.
 """

 import json
@@ -24,7 +23,7 @@ from werkzeug.exceptions import Forbidden, Unauthorized
 from controllers.service_api.app import app as app_controller
 from controllers.service_api.app.app import AppInfoApi, AppMetaApi, AppParameterApi
 from controllers.service_api.app.error import AgentNotPublishedError, AppUnavailableError
-from core.app.apps.agent_app.errors import AgentAppNotPublishedError
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole, TenantStatus
 from models.base import TypeBase
 from models.enums import EndUserType
@@ -41,7 +40,7 @@ from models.model import (
    TagType,
 )
 from models.workflow import Workflow, WorkflowType
-from services.app_definition_query_service import AppDefinitionUnavailableError
+from services.app_definition_query_service import AppDefinitionNotPublishedError, AppDefinitionUnavailableError


@dataclass(frozen=True)
@@ -139,7 +138,6 @@ def app_db(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[A
    registry = scoped_session(maker)
    binding = _DatabaseBinding(engine=sqlite_engine, session=registry)
    monkeypatch.setattr("controllers.service_api.wraps.db", binding)
-    monkeypatch.setattr(app_controller, "db", binding)
    monkeypatch.setattr("models.model.db", binding)
    monkeypatch.setattr("models.account.db", binding)

@@ -270,79 +268,51 @@ def authenticated_controller(app_db: AppDatabase, monkeypatch: pytest.MonkeyPatc
    return app_db


-@pytest.mark.usefixtures("authenticated_controller")
-def test_get_parameters_for_persisted_chat_config(flask_app: Flask) -> None:
-    with flask_app.test_request_context("/parameters", headers={"Authorization": "Bearer token"}):
-        response = AppParameterApi().get()
-
-    assert response["opening_statement"] == "Hello"
-    assert response["suggested_questions"] == ["Question?"]
-    assert response["user_input_form"] == [{"text-input": {"label": "Name", "variable": "name", "required": True}}]
-
-
-def test_get_parameters_for_persisted_workflow(flask_app: Flask, authenticated_controller: AppDatabase) -> None:
-    authenticated_controller.update_app(mode=AppMode.WORKFLOW, app_model_config_id=None)
-
-    with flask_app.test_request_context("/parameters", headers={"Authorization": "Bearer token"}):
-        response = AppParameterApi().get()
-
-    assert response["user_input_form"] == []
-    assert response["suggested_questions"] == []
-
-
-def test_get_parameters_for_agent_uses_persisted_app(
+def test_get_parameters_queries_authenticated_app(
    flask_app: Flask, authenticated_controller: AppDatabase, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    authenticated_controller.update_app(mode=AppMode.AGENT, app_model_config_id=None, workflow_id=None)
-    user_input_form = [{"text-input": {"label": "Topic", "variable": "topic", "required": True}}]
-    agent_parameters = Mock(
-        return_value=(
-            {"opening_statement": "Hi from Agent"},
-            user_input_form,
-        )
+    app_definitions = Mock()
+    app_definitions.get_public_parameters.return_value = get_parameters_from_feature_dict(
+        features_dict={"opening_statement": "Hello"},
+        user_input_form=[],
    )
-    monkeypatch.setattr(app_controller, "_get_agent_app_feature_dict_and_user_input_form", agent_parameters)
-
-    with flask_app.test_request_context("/parameters", headers={"Authorization": "Bearer token"}):
-        response = AppParameterApi().get()
-
-    assert response["opening_statement"] == "Hi from Agent"
-    assert response["user_input_form"] == user_input_form
-    agent_parameters.assert_called_once()
-    (app_model,) = agent_parameters.call_args.args
-    assert app_model.id == authenticated_controller.app_id
-    assert agent_parameters.call_args.kwargs["session"] is authenticated_controller.registry()
-
-
-def test_unpublished_agent_raises_friendly_error(
-    flask_app: Flask, authenticated_controller: AppDatabase, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    authenticated_controller.update_app(mode=AppMode.AGENT)
    monkeypatch.setattr(
        app_controller,
-        "get_published_agent_app_feature_dict_and_user_input_form",
-        Mock(side_effect=AgentAppNotPublishedError("not published")),
+        "application_services",
+        Mock(return_value=SimpleNamespace(app_definitions=app_definitions)),
    )

    with flask_app.test_request_context("/parameters", headers={"Authorization": "Bearer token"}):
-        with pytest.raises(AgentNotPublishedError):
-            AppParameterApi().get()
+        response = AppParameterApi().get()
+
+    app_definitions.get_public_parameters.assert_called_once_with(authenticated_controller.app_id)
+    assert response["opening_statement"] == "Hello"


@pytest.mark.parametrize(
-    ("mode", "field"),
-    [(AppMode.CHAT, "app_model_config_id"), (AppMode.WORKFLOW, "workflow_id")],
+    ("service_error", "http_error"),
+    [
+        pytest.param(AppDefinitionNotPublishedError(), AgentNotPublishedError, id="not-published"),
+        pytest.param(AppDefinitionUnavailableError(), AppUnavailableError, id="unavailable"),
+    ],
 )
-def test_parameters_reject_missing_persisted_configuration(
+@pytest.mark.usefixtures("authenticated_controller")
+def test_get_parameters_maps_query_errors(
    flask_app: Flask,
-    authenticated_controller: AppDatabase,
-    mode: AppMode,
-    field: str,
+    monkeypatch: pytest.MonkeyPatch,
+    service_error: Exception,
+    http_error: type[Exception],
 ) -> None:
-    authenticated_controller.update_app(mode=mode, **{field: None})
+    app_definitions = Mock()
+    app_definitions.get_public_parameters.side_effect = service_error
+    monkeypatch.setattr(
+        app_controller,
+        "application_services",
+        Mock(return_value=SimpleNamespace(app_definitions=app_definitions)),
+    )

    with flask_app.test_request_context("/parameters", headers={"Authorization": "Bearer token"}):
-        with pytest.raises(AppUnavailableError):
+        with pytest.raises(http_error):
            AppParameterApi().get()


@@ -10,106 +10,52 @@ from flask import Flask

 from controllers.web.app import AppAccessMode, AppMeta, AppParameterApi, AppWebAuthPermission
 from controllers.web.error import AgentNotPublishedError, AppUnavailableError
-from core.app.apps.agent_app.errors import AgentAppNotPublishedError
-from services.app_definition_query_service import AppDefinitionUnavailableError
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from services.app_definition_query_service import AppDefinitionNotPublishedError, AppDefinitionUnavailableError


 # ---------------------------------------------------------------------------
 # AppParameterApi
 # ---------------------------------------------------------------------------
 class TestAppParameterApi:
-    def test_advanced_chat_mode_uses_workflow(self, app: Flask) -> None:
-        features_dict = {"opening_statement": "Hello"}
-        workflow = SimpleNamespace(
-            features_dict=features_dict,
-            user_input_form=lambda to_old_structure=False: [],
-        )
-        app_model = SimpleNamespace(
-            mode="advanced-chat",
-            workflow_with_session=lambda *, session: workflow,
+    @patch("controllers.web.app.application_services")
+    def test_get_returns_public_parameters(self, application_services: MagicMock, app: Flask) -> None:
+        app_definitions = MagicMock()
+        app_definitions.get_public_parameters.return_value = get_parameters_from_feature_dict(
+            features_dict={"opening_statement": "Hello"},
+            user_input_form=[],
        )
+        application_services.return_value = SimpleNamespace(app_definitions=app_definitions)
+        app_model = SimpleNamespace(id="app-1")

-        with (
-            app.test_request_context("/parameters"),
-            patch("controllers.web.app.get_parameters_from_feature_dict", return_value={}) as mock_params,
-            patch("controllers.web.app.fields.Parameters") as mock_fields,
-        ):
-            mock_fields.model_validate.return_value.model_dump.return_value = {"result": "ok"}
+        with app.test_request_context("/parameters"):
            result = AppParameterApi().get(app_model, SimpleNamespace())

-        mock_params.assert_called_once_with(features_dict=features_dict, user_input_form=[])
-        assert result == {"result": "ok"}
+        assert result["opening_statement"] == "Hello"
+        app_definitions.get_public_parameters.assert_called_once_with("app-1")

-    def test_workflow_mode_uses_workflow(self, app: Flask) -> None:
-        features_dict = {}
-        workflow = SimpleNamespace(
-            features_dict=features_dict,
-            user_input_form=lambda to_old_structure=False: [{"var": "x"}],
-        )
-        app_model = SimpleNamespace(
-            mode="workflow",
-            workflow_with_session=lambda *, session: workflow,
-        )
+    @pytest.mark.parametrize(
+        ("service_error", "http_error"),
+        [
+            pytest.param(AppDefinitionNotPublishedError(), AgentNotPublishedError, id="not-published"),
+            pytest.param(AppDefinitionUnavailableError(), AppUnavailableError, id="unavailable"),
+        ],
+    )
+    @patch("controllers.web.app.application_services")
+    def test_get_maps_query_errors(
+        self,
+        application_services: MagicMock,
+        service_error: Exception,
+        http_error: type[Exception],
+        app: Flask,
+    ) -> None:
+        app_definitions = MagicMock()
+        app_definitions.get_public_parameters.side_effect = service_error
+        application_services.return_value = SimpleNamespace(app_definitions=app_definitions)

-        with (
-            app.test_request_context("/parameters"),
-            patch("controllers.web.app.get_parameters_from_feature_dict", return_value={}) as mock_params,
-            patch("controllers.web.app.fields.Parameters") as mock_fields,
-        ):
-            mock_fields.model_validate.return_value.model_dump.return_value = {}
-            AppParameterApi().get(app_model, SimpleNamespace())
-
-        mock_params.assert_called_once_with(features_dict=features_dict, user_input_form=[{"var": "x"}])
-
-    def test_advanced_chat_mode_no_workflow_raises(self, app: Flask) -> None:
-        app_model = SimpleNamespace(
-            mode="advanced-chat",
-            workflow_with_session=lambda *, session: None,
-        )
        with app.test_request_context("/parameters"):
-            with pytest.raises(AppUnavailableError):
-                AppParameterApi().get(app_model, SimpleNamespace())
-
-    def test_standard_mode_uses_app_model_config(self, app: Flask) -> None:
-        config = SimpleNamespace(to_dict=lambda **_kwargs: {"user_input_form": [{"var": "y"}], "key": "val"})
-        app_model = SimpleNamespace(
-            id="app-1",
-            mode="chat",
-            app_model_config_with_session=lambda *, session: config,
-        )
-
-        with (
-            app.test_request_context("/parameters"),
-            patch("controllers.web.app.get_parameters_from_feature_dict", return_value={}) as mock_params,
-            patch("controllers.web.app.fields.Parameters") as mock_fields,
-            patch("controllers.web.app.load_annotation_reply_config", return_value={"enabled": False}),
-        ):
-            mock_fields.model_validate.return_value.model_dump.return_value = {}
-            AppParameterApi().get(app_model, SimpleNamespace())
-
-        call_kwargs = mock_params.call_args
-        assert call_kwargs.kwargs["user_input_form"] == [{"var": "y"}]
-
-    def test_standard_mode_no_config_raises(self, app: Flask) -> None:
-        app_model = SimpleNamespace(
-            mode="chat",
-            app_model_config_with_session=lambda *, session: None,
-        )
-        with app.test_request_context("/parameters"):
-            with pytest.raises(AppUnavailableError):
-                AppParameterApi().get(app_model, SimpleNamespace())
-
-    def test_agent_mode_unpublished_raises_friendly_error(self, app: Flask) -> None:
-        app_model = SimpleNamespace(mode="agent")
-        with (
-            app.test_request_context("/parameters"),
-            patch(
-                "controllers.web.app.get_published_agent_app_feature_dict_and_user_input_form",
-                side_effect=AgentAppNotPublishedError("Agent has not been published"),
-            ),
-        ):
-            with pytest.raises(AgentNotPublishedError):
-                AppParameterApi().get(app_model, SimpleNamespace())
+            with pytest.raises(http_error):
+                AppParameterApi().get(SimpleNamespace(id="app-1"), SimpleNamespace())


 # ---------------------------------------------------------------------------
@@ -28,7 +28,7 @@ project-excludes = [
  "configs/test_dify_config.py",
  "configs/test_env_consistency.py",
  "configs/test_nacos_http_client.py",
-  "controllers/common/test_agent_app_parameters.py",
+  "repositories/test_app_definition_query_repository_agent.py",
  "controllers/common/test_app_access.py",
  "controllers/common/test_errors.py",
  "controllers/common/test_fields.py",
@@ -145,6 +145,44 @@ def test_get_published_parameter_config_returns_none_for_app_without_published_m
    assert repository.get_published_parameter_config(_APP_ID) is None


+def test_get_published_parameter_config_preserves_legacy_agent_model_config(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    with sqlite_session_factory() as session:
+        app = _persist_app(session, mode=AppMode.AGENT)
+        app_model_config = AppModelConfig(app_id=app.id, opening_statement="Legacy Agent config")
+        session.add(app_model_config)
+        session.flush()
+        app.app_model_config_id = app_model_config.id
+        session.commit()
+
+    result = AppDefinitionQueryRepository(session_factory=sqlite_session_factory).get_published_parameter_config(
+        _APP_ID
+    )
+
+    assert result is not None
+    assert result.features_dict["opening_statement"] == "Legacy Agent config"
+
+
+def test_get_public_parameter_config_reuses_standard_projection(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    with sqlite_session_factory() as session:
+        app = _persist_app(session)
+        app_model_config = AppModelConfig(app_id=app.id, opening_statement="Public config")
+        session.add(app_model_config)
+        session.flush()
+        app.app_model_config_id = app_model_config.id
+        session.commit()
+
+    result = AppDefinitionQueryRepository(session_factory=sqlite_session_factory).get_published_parameter_config(
+        _APP_ID, public_runtime=True
+    )
+
+    assert result is not None
+    assert result.features_dict["opening_statement"] == "Public config"
+
+
 def _tool(provider_type: str, provider_id: str, tool_name: str) -> dict[str, object]:
    return {
        "provider_type": provider_type,
@@ -3,9 +3,9 @@ from typing import Any
 from uuid import NAMESPACE_URL, uuid5

 import pytest
-from sqlalchemy.orm import Session
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker

-from controllers.common.agent_app_parameters import get_published_agent_app_feature_dict_and_user_input_form
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
 from models.agent import (
@@ -17,7 +17,11 @@ from models.agent import (
    AgentSource,
    AgentStatus,
 )
-from models.model import AppAnnotationSetting
+from models.model import App, AppAnnotationSetting, AppMode
+from repositories.app_definition_query_repository import (
+    AppDefinitionQueryRepository,
+    _get_public_agent_parameter_config,
+)


 def _stable_uuid(value: str) -> str:
@@ -28,7 +32,19 @@ def _app_model(*, tenant_id: str, bound_agent_id: str | None, app_model_config:
    return SimpleNamespace(
        id=_stable_uuid(f"app:{tenant_id}"),
        tenant_id=tenant_id,
-        bound_agent_id=bound_agent_id,
+        agent_app_binding_with_session=lambda *, session: (
+            session.scalar(
+                select(Agent)
+                .where(
+                    Agent.id == bound_agent_id,
+                    Agent.tenant_id == tenant_id,
+                    Agent.status == AgentStatus.ACTIVE,
+                )
+                .limit(1)
+            )
+            if bound_agent_id
+            else None
+        ),
        app_model_config_with_session=lambda *, session: app_model_config,
    )

@@ -40,6 +56,7 @@ def _persist_agent(
    agent_id: str,
    active_config_snapshot_id: str | None,
    active_config_is_published: bool,
+    app_id: str | None = None,
 ) -> Agent:
    agent = Agent(
        id=agent_id,
@@ -50,6 +67,7 @@ def _persist_agent(
        status=AgentStatus.ACTIVE,
        active_config_snapshot_id=active_config_snapshot_id,
        active_config_is_published=active_config_is_published,
+        app_id=app_id,
    )
    session.add(agent)
    session.commit()
@@ -107,6 +125,54 @@ def _persist_publish_revision(
        session.commit()


+def test_get_public_parameter_config_loads_agent_snapshot(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    tenant_id = _stable_uuid("tenant:repository")
+    app_id = _stable_uuid("app:repository")
+    agent_id = _stable_uuid("agent:repository")
+    snapshot_id = _stable_uuid("snapshot:repository")
+    with sqlite_session_factory() as session:
+        session.add(
+            App(
+                id=app_id,
+                tenant_id=tenant_id,
+                name="Agent App",
+                description="",
+                mode=AppMode.AGENT,
+                icon_type=None,
+                icon=None,
+                icon_background=None,
+                enable_site=True,
+                enable_api=True,
+                max_active_requests=None,
+            )
+        )
+        session.commit()
+        _persist_agent(
+            session,
+            tenant_id=tenant_id,
+            agent_id=agent_id,
+            active_config_snapshot_id=snapshot_id,
+            active_config_is_published=True,
+            app_id=app_id,
+        )
+        _persist_snapshot(
+            session,
+            snapshot_id=snapshot_id,
+            tenant_id=tenant_id,
+            agent_id=agent_id,
+            config_snapshot={"app_variables": [{"name": "topic", "type": "string", "required": True}]},
+        )
+
+    result = AppDefinitionQueryRepository(session_factory=sqlite_session_factory).get_published_parameter_config(
+        app_id, public_runtime=True
+    )
+
+    assert result is not None
+    assert result.user_input_form == [{"text-input": {"label": "topic", "variable": "topic", "required": True}}]
+
+
@pytest.mark.parametrize(
    "sqlite_session",
    [(Agent, AgentConfigSnapshot, AgentConfigRevision, AppAnnotationSetting)],
@@ -157,7 +223,7 @@ def test_published_agent_app_parameters_use_soul_file_upload(sqlite_session: Ses
        },
    )

-    features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
+    features_dict, user_input_form = _get_public_agent_parameter_config(
        app_model,
        session=sqlite_session,
    )
@@ -181,50 +247,24 @@ def test_published_agent_app_parameters_requires_bound_agent(sqlite_session: Ses
    app_model = _app_model(tenant_id=tenant_id, bound_agent_id=None)

    with pytest.raises(AgentAppGeneratorError, match="no bound Agent"):
-        get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
+        _get_public_agent_parameter_config(app_model, session=sqlite_session)


@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot, AgentConfigRevision)], indirect=True)
-def test_published_agent_app_parameters_requires_existing_active_agent(sqlite_session: Session):
-    requested_tenant_id = _stable_uuid("tenant:requested")
-    agent_id = _stable_uuid("agent:cross-tenant")
-    app_model = _app_model(tenant_id=requested_tenant_id, bound_agent_id=agent_id)
-    _persist_agent(
-        sqlite_session,
-        tenant_id=_stable_uuid("tenant:other"),
-        agent_id=agent_id,
-        active_config_snapshot_id=None,
-        active_config_is_published=False,
-    )
-
-    with pytest.raises(AgentAppGeneratorError, match="no bound Agent"):
-        get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
-
-
-@pytest.mark.parametrize(
-    "active_config_is_published",
-    [
-        True,
-        False,
-    ],
-)
-@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot, AgentConfigRevision)], indirect=True)
-def test_published_agent_app_parameters_requires_published_agent(
-    active_config_is_published: bool, sqlite_session: Session
-):
-    tenant_id = _stable_uuid(f"tenant:published:{active_config_is_published}")
-    agent_id = _stable_uuid(f"agent:published:{active_config_is_published}")
+def test_published_agent_app_parameters_requires_published_agent(sqlite_session: Session):
+    tenant_id = _stable_uuid("tenant:published")
+    agent_id = _stable_uuid("agent:published")
    app_model = _app_model(tenant_id=tenant_id, bound_agent_id=agent_id)
    _persist_agent(
        sqlite_session,
        tenant_id=tenant_id,
        agent_id=agent_id,
        active_config_snapshot_id=None,
-        active_config_is_published=active_config_is_published,
+        active_config_is_published=True,
    )

    with pytest.raises(AgentAppNotPublishedError, match="not been published"):
-        get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
+        _get_public_agent_parameter_config(app_model, session=sqlite_session)


@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot, AgentConfigRevision)], indirect=True)
@@ -248,7 +288,7 @@ def test_published_agent_app_parameters_allows_unpublished_draft_with_active_sna
        config_snapshot={},
    )

-    features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
+    features_dict, user_input_form = _get_public_agent_parameter_config(
        app_model,
        session=sqlite_session,
    )
@@ -280,7 +320,7 @@ def test_published_agent_app_parameters_rejects_seeded_unpublished_snapshot(sqli
    )

    with pytest.raises(AgentAppNotPublishedError, match="not been published"):
-        get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
+        _get_public_agent_parameter_config(app_model, session=sqlite_session)


@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot, AgentConfigRevision)], indirect=True)
@@ -303,7 +343,7 @@ def test_published_agent_app_parameters_requires_published_snapshot(sqlite_sessi
    )

    with pytest.raises(AgentAppGeneratorError, match="published version not found"):
-        get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
+        _get_public_agent_parameter_config(app_model, session=sqlite_session)


@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot, AgentConfigRevision)], indirect=True)
@@ -327,7 +367,7 @@ def test_published_agent_app_parameters_allows_missing_legacy_app_model_config(s
        config_snapshot={},
    )

-    features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
+    features_dict, user_input_form = _get_public_agent_parameter_config(
        app_model,
        session=sqlite_session,
    )
@@ -3,7 +3,9 @@ from unittest.mock import MagicMock, create_autospec, patch
 import pytest

 import services.app_definition_query_service as module
+from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
 from services.app_definition_query_service import (
+    AppDefinitionNotPublishedError,
    AppDefinitionQuery,
    AppDefinitionQueryService,
    AppDefinitionUnavailableError,
@@ -53,6 +55,37 @@ def test_get_parameters_rejects_missing_config() -> None:
        service.get_parameters("app-1")


+def test_get_public_parameters_maps_public_config() -> None:
+    service, definitions = _service()
+    config = AppParameterConfig(
+        features_dict={"opening_statement": "Hello"},
+        user_input_form=[{"text-input": {"variable": "query"}}],
+    )
+    definitions.get_published_parameter_config.return_value = config
+    mapped = {"mapped": True}
+
+    with patch.object(module, "get_parameters_from_feature_dict", return_value=mapped):
+        result = service.get_public_parameters("app-1")
+
+    assert result is mapped
+    definitions.get_published_parameter_config.assert_called_once_with("app-1", public_runtime=True)
+
+
+@pytest.mark.parametrize(
+    ("source_error", "expected_error"),
+    [
+        pytest.param(AgentAppNotPublishedError("not published"), AppDefinitionNotPublishedError, id="not-published"),
+        pytest.param(AgentAppGeneratorError("unavailable"), AppDefinitionUnavailableError, id="unavailable"),
+    ],
+)
+def test_get_public_parameters_maps_agent_errors(source_error: Exception, expected_error: type[Exception]) -> None:
+    service, definitions = _service()
+    definitions.get_published_parameter_config.side_effect = source_error
+
+    with pytest.raises(expected_error):
+        service.get_public_parameters("app-1")
+
+
 def test_get_tool_icons_maps_builtin_and_api_icons() -> None:
    service, definitions = _service()
    definitions.get_tool_icon_sources.return_value = (