refactor(api): isolate trial app and workflow detail previews (#42659)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/.importlinter

@@ -589,6 +589,7 @@ name = App preview queries are framework and persistence neutral
 type = forbidden
 source_modules =
     services.app_preview_query_service
+    services.app_preview_details_service
 forbidden_modules =
     controllers
     extensions

api/controllers/console/__init__.py

@@ -1,5 +1,7 @@
+from collections import OrderedDict
 from http import HTTPStatus
 from importlib import import_module
+from typing import cast
 
 from flask import Blueprint, current_app, got_request_exception
 from flask_restx import Namespace
@@ -29,6 +31,11 @@ def _handle_active_workspace_required_error(error: ActiveWorkspaceRequiredError)
     }, status.value
 
 
+# Flask-RESTX dispatches in registration order; the shared Exception handler
+# must not shadow this Console-specific error mapping.
+cast(OrderedDict[type[Exception], object], api.error_handlers).move_to_end(ActiveWorkspaceRequiredError, last=False)
+
+
 console_ns = Namespace("console", description="Console management API operations", path="/")
 
 RESOURCE_MODULES = (

api/controllers/console/app/wraps.py

@@ -1,9 +1,8 @@
 """Controller decorators for console app resources.
 
 `get_app_model` still supports legacy handlers backed by Flask-SQLAlchemy's
-scoped session. Preview handlers compose `get_previewable_app_model` under
-`controllers.common.session.with_session`; preview admission finishes before
-the request Session loads the accepted App.
+scoped session. Public catalog previews use `app.preview_admission` and
+detached application queries.
 """
 
 from collections.abc import Callable
@@ -15,16 +14,13 @@ from sqlalchemy.orm import Session, scoped_session
 
 from controllers.common.session import with_session
 from controllers.console.app.error import AppNotFoundError
-from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models import App, AppMode
 from models.agent import AgentScope
-from services.app_service import AppService
 
 __all__ = [
     "get_app_model",
-    "get_previewable_app_model",
     "with_session",
 ]
 
@@ -56,13 +52,6 @@ def _load_app_model_from_scoped_session(app_id: str) -> App | None:
     return app_model
 
 
-def _load_previewable_app_model(session: Session, app_id: str) -> App | None:
-    """Load a normal App after preview admission completes outside the request Session."""
-    if not application_services().recommended_app_queries.is_previewable(app_id):
-        return None
-    return AppService.get_normal_app_by_id(app_id, session)
-
-
 def _get_injected_session(args: tuple[object, ...]) -> Session | None:
     """Return the request session inserted by `with_session`, if this handler has been migrated."""
     if len(args) < 2:
@@ -148,75 +137,3 @@ def get_app_model[**P, R](
         return decorator
     else:
         return decorator(view)
-
-
-@overload
-def get_previewable_app_model[**P, R](
-    view: Callable[P, R],
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R]: ...
-
-
-@overload
-def get_previewable_app_model[**P, R](
-    view: None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[[Callable[P, R]], Callable[P, R]]: ...
-
-
-def get_previewable_app_model[**P, R](
-    view: Callable[P, R] | None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
-    """Inject an App authorized for read-only template preview.
-
-    Preview reads accept either an explicit TrialApp registration or membership
-    in the recommended catalog. This does not grant trial execution, which is
-    separately protected by TrialAppResource's feature, registration, and quota
-    checks.
-    """
-
-    def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
-        @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
-            if not kwargs.get("app_id"):
-                raise ValueError("missing app_id in path parameters")
-
-            app_id = kwargs.get("app_id")
-            app_id = str(app_id)
-
-            del kwargs["app_id"]
-
-            session = _get_injected_session(args)
-            if session is None:
-                raise RuntimeError("get_previewable_app_model requires @with_session")
-            app_model = _load_previewable_app_model(session, app_id)
-
-            if not app_model:
-                raise AppNotFoundError()
-
-            app_mode = AppMode.value_of(app_model.mode)
-
-            if mode is not None:
-                if isinstance(mode, list):
-                    modes = mode
-                else:
-                    modes = [mode]
-
-                if app_mode not in modes:
-                    mode_values = {m.value for m in modes}
-                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
-
-            kwargs["app_model"] = app_model
-
-            return view_func(*args, **kwargs)
-
-        return decorated_view
-
-    if view is None:
-        return decorator
-    else:
-        return decorator(view)

api/controllers/console/explore/trial.py

@@ -1,5 +1,6 @@
 import logging
-from dataclasses import dataclass
+from collections.abc import Mapping
+from dataclasses import asdict
 from datetime import datetime
 from typing import Any, Literal
 from uuid import UUID
@@ -7,10 +8,10 @@ from uuid import UUID
 from flask import Response, request
 from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, field_validator
-from sqlalchemy.orm import Session
 from werkzeug.exceptions import InternalServerError, NotFound, Unauthorized
 
 import services
+from configs import dify_config
 from controllers.common.fields import (
     AudioBinaryResponse,
     AudioTranscriptResponse,
@@ -41,7 +42,6 @@ from controllers.console.app.error import (
     UnsupportedAudioTypeError,
 )
 from controllers.console.app.preview_admission import get_preview_app
-from controllers.console.app.wraps import get_previewable_app_model, with_session
 from controllers.console.explore.error import (
     AppPreviewOwnerUnavailableError as AppPreviewOwnerUnavailableHttpError,
 )
@@ -79,10 +79,10 @@ from graphon.model_runtime.errors.invoke import InvokeError
 from graphon.variables import SecretVariable, VariableBase
 from libs import helper
 from libs.helper import dump_response, to_timestamp, uuid_value
+from libs.url_utils import normalize_api_base_url
 from machinery.context import RequestContext
 from models import Account
 from models.enums import CreatorUserRole
-from models.workflow import Workflow
 from services.account_errors import AccountNotFoundError
 from services.app_definition_query_service import AppDefinitionUnavailableError
 from services.app_preview_query_service import (
@@ -92,7 +92,6 @@ from services.app_preview_query_service import (
     AppPreviewUnavailableError,
 )
 from services.app_ref_service import AppRefService
-from services.app_service import AppResponseView, AppService
 from services.audio_service import AudioService
 from services.errors.audio import (
     AudioTooLargeServiceError,
@@ -414,7 +413,12 @@ class TrialWorkflowResponse(ResponseModel):
 
         result: list[Any] = []
         for item in value:
-            if isinstance(item, SecretVariable):
+            if isinstance(item, Mapping):
+                serialized = dict(item)
+                if serialized.get("value_type") == "secret":
+                    serialized["value"] = encrypter.full_mask_token()
+                result.append(serialized)
+            elif isinstance(item, SecretVariable):
                 serialized = item.model_dump(mode="json")
                 serialized["value"] = encrypter.full_mask_token()
                 result.append(serialized)
@@ -427,27 +431,6 @@ class TrialWorkflowResponse(ResponseModel):
         return result
 
 
-@dataclass(frozen=True)
-class TrialWorkflowResponseSource:
-    workflow: Workflow
-    session: Session
-
-    @property
-    def created_by_account(self) -> Account | None:
-        return self.workflow.get_created_by_account(session=self.session)
-
-    @property
-    def updated_by_account(self) -> Account | None:
-        return self.workflow.get_updated_by_account(session=self.session)
-
-    @property
-    def tool_published(self) -> bool:
-        return self.workflow.get_tool_published(session=self.session)
-
-    def __getattr__(self, name: str) -> Any:
-        return getattr(self.workflow, name)  # guard-ignore: no-new-getattr -- delegates model fields
-
-
 register_schema_models(
     console_ns,
     WorkflowRunRequest,
@@ -849,37 +832,46 @@ class TrialAppParameterApi(Resource):
 
 class AppApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[TrialAppDetailResponse.__name__])
-    @with_session(write=False)
-    @get_previewable_app_model(None)
-    def get(self, session: Session, app_model):
-        """Get app detail"""
+    @console_account_admission()
+    @get_preview_app
+    def get(self, app: AppPreviewRef, request_context: RequestContext) -> dict[str, object]:
+        """Get app detail using the viewer's workspace for tool configuration."""
+        try:
+            detail = application_services().app_preview_details.get_detail(
+                app=app,
+                account_id=request_context.account_id,
+                active_workspace_id=request_context.active_workspace_id,
+            )
+        except AppPreviewUnavailableError as error:
+            raise AppNotFoundError() from error
+        except AppDefinitionUnavailableError as error:
+            raise AppUnavailableError() from error
+        except AccountNotFoundError as error:
+            raise Unauthorized("Account no longer exists.") from error
+        except AppPreviewSiteUnavailableError as error:
+            raise AppPreviewSiteUnavailableHttpError(str(error)) from error
 
-        app_service = AppService()
-        app_model = app_service.get_app(app_model, session=session)
-
-        return TrialAppDetailResponse.model_validate(
-            AppResponseView(app_model, session=session),
-            from_attributes=True,
-        ).model_dump(mode="json")
+        source = asdict(detail)
+        source["api_base_url"] = normalize_api_base_url(dify_config.SERVICE_API_URL or request.host_url.rstrip("/"))
+        source["site"] = {
+            **asdict(detail.site),
+            "app_base_url": dify_config.APP_WEB_URL or request.url_root.rstrip("/"),
+        }
+        return dump_response(TrialAppDetailResponse, source)
 
 
 class AppWorkflowApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[TrialWorkflowResponse.__name__])
-    @with_session(write=False)
-    @get_previewable_app_model(None)
-    def get(self, session: Session, app_model):
-        """Get workflow detail"""
-        if not app_model.workflow_id:
-            raise AppUnavailableError()
-
-        workflow = app_model.workflow_with_session(session=session)
-        if workflow is None:
-            raise AppUnavailableError()
-
-        return TrialWorkflowResponse.model_validate(
-            TrialWorkflowResponseSource(workflow=workflow, session=session),
-            from_attributes=True,
-        ).model_dump(mode="json")
+    @get_preview_app
+    def get(self, app: AppPreviewRef) -> dict[str, object]:
+        """Get a detached workflow definition after catalog preview admission."""
+        try:
+            workflow = application_services().app_preview_details.get_workflow(app=app)
+        except AppPreviewUnavailableError as error:
+            raise AppNotFoundError() from error
+        except AppDefinitionUnavailableError as error:
+            raise AppUnavailableError() from error
+        return dump_response(TrialWorkflowResponse, workflow)
 
 
 class DatasetListApi(Resource):

api/core/agent/tool_configuration.py

@@ -0,0 +1,52 @@
+"""Build agent tool configuration for responses without changing stored config."""
+
+import logging
+from collections.abc import Mapping
+from copy import deepcopy
+from typing import cast
+
+from pydantic import JsonValue, ValidationError
+
+from core.agent.entities import AgentToolEntity
+from core.tools.tool_manager import ToolManager
+from core.tools.utils.configuration import ToolParameterConfigurationManager
+
+logger = logging.getLogger(__name__)
+
+
+def mask_agent_tool_parameters(
+    *, agent_mode: Mapping[str, JsonValue], app_id: str, tenant_id: str, user_id: str
+) -> dict[str, JsonValue]:
+    """Return a response copy, exposing parameters only after masking succeeds."""
+    result = deepcopy(dict(agent_mode))
+    for tool in cast(list[JsonValue], result.get("tools") or []):
+        if not isinstance(tool, dict):
+            continue
+        unmasked_tool = dict(tool)
+        if "tool_parameters" in tool:
+            tool["tool_parameters"] = {}
+        if len(unmasked_tool) <= 3:
+            continue
+        try:
+            agent_tool = AgentToolEntity.model_validate(unmasked_tool)
+        except ValidationError:
+            continue
+        try:
+            tool_runtime = ToolManager.get_agent_tool_runtime(
+                tenant_id=tenant_id, app_id=app_id, agent_tool=agent_tool, user_id=user_id
+            )
+            manager = ToolParameterConfigurationManager(
+                tenant_id=tenant_id,
+                tool_runtime=tool_runtime,
+                provider_name=agent_tool.provider_id,
+                provider_type=agent_tool.provider_type,
+                identity_id=f"AGENT.{app_id}",
+            )
+            if agent_tool.tool_parameters:
+                parameters = manager.decrypt_tool_parameters(agent_tool.tool_parameters)
+                tool["tool_parameters"] = manager.mask_tool_parameters(parameters or {})
+            else:
+                tool["tool_parameters"] = {}
+        except Exception:
+            logger.exception("Failed to mask agent tool parameters for tool %s", agent_tool.tool_name)
+    return result

api/core/workflow/environment_variables.py

@@ -0,0 +1,33 @@
+"""Load persisted workflow environment variables without requiring an ORM model."""
+
+import json
+from typing import cast
+
+from core.helper import encrypter
+from core.workflow.llm_environment_variable import LLMEnvironmentVariable
+from factories import variable_factory
+from graphon.variables import FloatVariable, IntegerVariable, SecretVariable, StringVariable, VariableBase
+
+
+def load_environment_variables(
+    *, tenant_id: str | None, serialized_variables: str | None
+) -> list[StringVariable | IntegerVariable | FloatVariable | SecretVariable | LLMEnvironmentVariable]:
+    # Use the workflow owner to avoid relying on request user in background threads.
+    if not tenant_id:
+        return []
+
+    variables = cast(dict[str, dict[str, object]], json.loads(serialized_variables or "{}"))
+    results = [variable_factory.build_environment_variable_from_mapping(value) for value in variables.values()]
+
+    def decrypt_func(
+        var: VariableBase,
+    ) -> StringVariable | IntegerVariable | FloatVariable | SecretVariable | LLMEnvironmentVariable:
+        match var:
+            case SecretVariable():
+                return var.model_copy(update={"value": encrypter.decrypt_token(tenant_id=tenant_id, token=var.value)})
+            case StringVariable() | IntegerVariable() | FloatVariable() | LLMEnvironmentVariable():
+                return var
+            case _:
+                raise AssertionError(f"Unexpected variable type for environment variable: {type(var)}")
+
+    return [decrypt_func(var) for var in results]

api/extensions/ext_application_services.py

@@ -137,6 +137,8 @@ from services.account_password_hasher import DefaultAccountPasswordHasher
 from services.account_password_service import AccountPasswordService
 from services.account_profile_service import AccountProfileService
 from services.app_definition_query_service import AppDefinitionQueryService
+from services.app_preview_details_adapters import AppPreviewDetailsRuntime
+from services.app_preview_details_service import AppPreviewDetails
 from services.app_preview_query_service import AppPreviewQueryService
 from services.app_site_service import AppSiteService
 from services.app_statistic_query import AppStatisticQuery
@@ -262,6 +264,7 @@ class ApplicationServices:
     accounts: AccountServices
     account_activation: AccountActivationService
     app_definitions: AppDefinitionQueryService
+    app_preview_details: AppPreviewDetails
     app_previews: AppPreviewQueryService
     app_sites: AppSiteService
     app_statistics: AppStatisticQuery
@@ -436,6 +439,7 @@ def build_application_services(
     installation_state = InstallationStateRepository(session_factory=database_client)
     data_source_api_key_auth_bindings = SQLAlchemyDataSourceApiKeyAuthBindingRepository(session_factory=database_client)
     app_definition_repository = AppDefinitionQueryRepository(session_factory=database_client)
+    app_preview_repository = AppPreviewQueryRepository(session_factory=database_client)
     feature_gateway = FeatureServiceGateway()
     accounts = SQLAlchemyAccountRepository(session_factory=database_client)
     integrations = SQLAlchemyAccountIntegrationRepository(session_factory=database_client)
@@ -626,8 +630,9 @@ def build_application_services(
                 dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
             ),
         ),
+        app_preview_details=AppPreviewDetailsRuntime(details=app_preview_repository),
         app_previews=AppPreviewQueryService(
-            apps=AppPreviewQueryRepository(session_factory=database_client),
+            apps=app_preview_repository,
             is_previewable=recommended_app_queries.is_previewable,
         ),
         app_sites=AppSiteService(

api/models/workflow.py

@@ -24,6 +24,7 @@ from sqlalchemy.orm import Mapped, mapped_column
 from typing_extensions import deprecated
 
 from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
+from core.workflow.environment_variables import load_environment_variables
 from core.workflow.human_input_adapter import adapt_node_config_for_graph
 from core.workflow.llm_environment_variable import LLMEnvironmentVariable, dump_environment_variable
 from core.workflow.nodes.human_input.pause_reason import (
@@ -589,36 +590,7 @@ class Workflow(Base):  # bug
     def environment_variables(
         self,
     ) -> Sequence[StringVariable | IntegerVariable | FloatVariable | SecretVariable | LLMEnvironmentVariable]:
-        # Use workflow.tenant_id to avoid relying on request user in background threads
-        tenant_id = self.tenant_id
-
-        if not tenant_id:
-            return []
-
-        environment_variables_dict = cast(SerializedWorkflowVariables, json.loads(self._environment_variables or "{}"))
-        results = [
-            variable_factory.build_environment_variable_from_mapping(v) for v in environment_variables_dict.values()
-        ]
-
-        # decrypt secret variables value
-        def decrypt_func(
-            var: VariableBase,
-        ) -> StringVariable | IntegerVariable | FloatVariable | SecretVariable | LLMEnvironmentVariable:
-            match var:
-                case SecretVariable():
-                    return var.model_copy(
-                        update={"value": encrypter.decrypt_token(tenant_id=tenant_id, token=var.value)}
-                    )
-                case StringVariable() | IntegerVariable() | FloatVariable() | LLMEnvironmentVariable():
-                    return var
-                case _:
-                    # Other variable types are not supported for environment variables
-                    raise AssertionError(f"Unexpected variable type for environment variable: {type(var)}")
-
-        decrypted_results: list[
-            SecretVariable | StringVariable | IntegerVariable | FloatVariable | LLMEnvironmentVariable
-        ] = [decrypt_func(var) for var in results]
-        return decrypted_results
+        return load_environment_variables(tenant_id=self.tenant_id, serialized_variables=self._environment_variables)
 
     @environment_variables.setter
     def environment_variables(self, value: Sequence[VariableBase]):

api/openapi/markdown/console-openapi.md

@@ -9351,7 +9351,7 @@ Remove one or more tag bindings from a target.
 | 200 | Success | **application/json**: [TagResponse](#tagresponse)<br> |
 
 ### [GET] /trial-apps/{app_id}
-**Get app detail**
+**Get app detail using the viewer's workspace for tool configuration**
 
 #### Parameters
 
@@ -9540,7 +9540,7 @@ Returns the site configuration for the application including theme, icons, and t
 | 200 | Success | **application/json**: [AudioBinaryResponse](#audiobinaryresponse)<br> |
 
 ### [GET] /trial-apps/{app_id}/workflows
-**Get workflow detail**
+**Get a detached workflow definition after catalog preview admission**
 
 #### Parameters
 

api/repositories/app_preview_query_repository.py

@@ -1,24 +1,46 @@
 """Load detached app-preview data without trial or account admission policy."""
 
-from collections.abc import Sequence
-from typing import override
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from typing import cast, override
 
+from pydantic import JsonValue
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
-from models.account import Tenant
+from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.utils.uuid_utils import is_valid_uuid
+from models.account import Account, Tenant
 from models.dataset import Dataset
-from models.model import App, Site
+from models.model import App, AppModelConfig, Site, load_annotation_reply_config
+from models.tools import ApiToolProvider
+from models.workflow import Workflow
 from repositories.app_definition_query_repository import map_site_configuration
+from services.account_errors import AccountNotFoundError
+from services.app_definition_query_service import AppDefinitionUnavailableError
+from services.app_preview_details_service import (
+    AppPreviewAccount,
+    AppPreviewAudit,
+    AppPreviewDetail,
+    AppPreviewDetailRecord,
+    AppPreviewDetailSite,
+    AppPreviewDetailsQuery,
+    AppPreviewObject,
+    AppPreviewTag,
+    AppPreviewWorkflow,
+    AppPreviewWorkflowRecord,
+)
 from services.app_preview_query_service import (
     AppPreviewDataset,
     AppPreviewQuery,
     AppPreviewRef,
     AppPreviewSite,
+    AppPreviewSiteUnavailableError,
+    AppPreviewUnavailableError,
 )
 
 
-class AppPreviewQueryRepository(AppPreviewQuery):
+class AppPreviewQueryRepository(AppPreviewQuery, AppPreviewDetailsQuery):
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._session_factory: sessionmaker[Session] = session_factory
 
@@ -77,3 +99,225 @@ class AppPreviewQueryRepository(AppPreviewQuery):
                 )
                 for dataset in datasets
             )
+
+    @override
+    def get_detail(self, *, app: AppPreviewRef, account_id: str) -> AppPreviewDetailRecord:
+        with self._session_factory(expire_on_commit=False) as session:
+            app_model = self._get_app(session=session, app=app)
+            account = session.get(Account, account_id)
+            if account is None:
+                raise AccountNotFoundError(f"Account {account_id} no longer exists")
+            model_config = self._get_model_config(session=session, app=app_model)
+            workflow = self._get_workflow(session=session, app=app_model)
+            agent_mode = model_config.agent_mode_dict if model_config is not None else None
+            tools = tuple(cast(Sequence[AppPreviewObject], agent_mode.get("tools", []))) if agent_mode else ()
+            existing_api_provider_ids = self._get_existing_api_provider_ids(
+                session=session, tenant_id=app.tenant_id, tools=tools
+            )
+            configuration = self._map_model_config(session=session, model_config=model_config)
+            site = session.scalar(select(Site).where(Site.app_id == app.app_id).limit(1))
+            if site is None:
+                raise AppPreviewSiteUnavailableError(f"Site for app {app.app_id} is unavailable")
+
+            detail = AppPreviewDetail(
+                id=app_model.id,
+                name=app_model.name,
+                description=app_model.description,
+                mode=app_model.mode,
+                icon_type=app_model.icon_type.value if app_model.icon_type is not None else None,
+                icon=app_model.icon,
+                icon_background=app_model.icon_background,
+                enable_site=app_model.enable_site,
+                enable_api=app_model.enable_api,
+                model_config=configuration,
+                workflow=self._map_workflow_summary(workflow),
+                use_icon_as_answer_icon=app_model.use_icon_as_answer_icon,
+                max_active_requests=app_model.max_active_requests,
+                created_by=app_model.created_by,
+                created_at=app_model.created_at,
+                updated_by=app_model.updated_by,
+                updated_at=app_model.updated_at,
+                deleted_tools=(),
+                tags=tuple(
+                    AppPreviewTag(id=tag.id, name=tag.name, type=tag.type)
+                    for tag in app_model.tags_with_session(session=session)
+                ),
+                site=self._map_site(site),
+            )
+
+            return AppPreviewDetailRecord(detail=detail, existing_api_provider_ids=existing_api_provider_ids)
+
+    @override
+    def get_workflow(self, *, app: AppPreviewRef) -> AppPreviewWorkflowRecord:
+        with self._session_factory() as session:
+            app_model = self._get_app(session=session, app=app)
+            workflow = self._get_workflow(session=session, app=app_model)
+            if workflow is None:
+                raise AppDefinitionUnavailableError(f"Workflow for app {app.app_id} is unavailable")
+            created_by = self._map_account(workflow.get_created_by_account(session=session))
+            updated_by = self._map_account(workflow.get_updated_by_account(session=session))
+            tool_published = workflow.get_tool_published(session=session)
+
+        # Legacy feature normalization mutates the model property. Keep that
+        # response-only conversion on the detached object after all queries finish.
+        detail = AppPreviewWorkflow(
+            id=workflow.id,
+            graph=cast(AppPreviewObject, workflow.graph_dict),
+            features=cast(AppPreviewObject, workflow.features_dict),
+            hash=workflow.unique_hash,
+            version=workflow.version,
+            marked_name=workflow.marked_name,
+            marked_comment=workflow.marked_comment,
+            created_by=created_by,
+            created_at=workflow.created_at,
+            updated_by=updated_by,
+            updated_at=workflow.updated_at,
+            tool_published=tool_published,
+            environment_variables=(),
+            conversation_variables=tuple(
+                cast(
+                    AppPreviewObject,
+                    {**variable.model_dump(mode="json"), "value_type": str(variable.value_type.exposed_type())},
+                )
+                for variable in workflow.conversation_variables
+            ),
+            rag_pipeline_variables=tuple(
+                cast(AppPreviewObject, variable) for variable in workflow.rag_pipeline_variables
+            ),
+        )
+        return AppPreviewWorkflowRecord(
+            workflow=detail,
+            tenant_id=workflow.tenant_id,
+            environment_variables_json=workflow._environment_variables,
+        )
+
+    @staticmethod
+    def _get_existing_api_provider_ids(
+        *, session: Session, tenant_id: str, tools: Sequence[AppPreviewObject]
+    ) -> frozenset[str]:
+        provider_ids = {
+            provider_id
+            for tool in tools
+            if len(tool) >= 4
+            and tool.get("provider_type") == ToolProviderType.API
+            and isinstance(provider_id := tool.get("provider_id"), str)
+            and is_valid_uuid(provider_id)
+        }
+        if not provider_ids:
+            return frozenset()
+        return frozenset(
+            session.scalars(
+                select(ApiToolProvider.id).where(
+                    ApiToolProvider.tenant_id == tenant_id, ApiToolProvider.id.in_(provider_ids)
+                )
+            )
+        )
+
+    @staticmethod
+    def _get_app(*, session: Session, app: AppPreviewRef) -> App:
+        app_model = session.scalar(
+            select(App).where(App.id == app.app_id, App.tenant_id == app.tenant_id, App.status == "normal")
+        )
+        if app_model is None:
+            raise AppPreviewUnavailableError(f"App {app.app_id} is no longer available in tenant {app.tenant_id}")
+        return app_model
+
+    @staticmethod
+    def _get_model_config(*, session: Session, app: App) -> AppModelConfig | None:
+        if app.app_model_config_id is None:
+            return None
+        config = session.scalar(
+            select(AppModelConfig).where(
+                AppModelConfig.id == app.app_model_config_id,
+                AppModelConfig.app_id == app.id,
+            )
+        )
+        if (
+            config is None
+            and session.scalar(select(AppModelConfig.id).where(AppModelConfig.id == app.app_model_config_id))
+            is not None
+        ):
+            raise AppDefinitionUnavailableError(
+                f"Model config {app.app_model_config_id} does not belong to app {app.id}"
+            )
+        return config
+
+    @staticmethod
+    def _get_workflow(*, session: Session, app: App) -> Workflow | None:
+        if app.workflow_id is None:
+            return None
+        workflow = session.scalar(
+            select(Workflow).where(
+                Workflow.id == app.workflow_id,
+                Workflow.app_id == app.id,
+                Workflow.tenant_id == app.tenant_id,
+            )
+        )
+        if workflow is None and session.scalar(select(Workflow.id).where(Workflow.id == app.workflow_id)) is not None:
+            raise AppDefinitionUnavailableError(f"Workflow {app.workflow_id} does not belong to app {app.id}")
+        return workflow
+
+    @staticmethod
+    def _map_model_config(
+        *, session: Session, model_config: AppModelConfig | None
+    ) -> dict[str, JsonValue | datetime] | None:
+        if model_config is None:
+            return None
+        configuration = model_config.to_dict(
+            annotation_reply=load_annotation_reply_config(session, model_config.app_id)
+        )
+        result: dict[str, JsonValue | datetime] = dict(cast(Mapping[str, JsonValue], configuration))
+        result.update(
+            created_by=model_config.created_by,
+            created_at=model_config.created_at,
+            updated_by=model_config.updated_by,
+            updated_at=model_config.updated_at,
+        )
+        return result
+
+    @staticmethod
+    def _map_workflow_summary(workflow: Workflow | None) -> AppPreviewAudit | None:
+        if workflow is None:
+            return None
+        return AppPreviewAudit(
+            id=workflow.id,
+            created_by=workflow.created_by,
+            created_at=workflow.created_at,
+            updated_by=workflow.updated_by,
+            updated_at=workflow.updated_at,
+        )
+
+    @staticmethod
+    def _map_account(account: Account | None) -> AppPreviewAccount | None:
+        if account is None:
+            return None
+        return AppPreviewAccount(id=account.id, name=account.name, email=account.email)
+
+    @staticmethod
+    def _map_site(site: Site) -> AppPreviewDetailSite:
+        return AppPreviewDetailSite(
+            code=site.code,
+            title=site.title,
+            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,
+            default_language=site.default_language,
+            chat_color_theme=site.chat_color_theme,
+            chat_color_theme_inverted=site.chat_color_theme_inverted,
+            customize_domain=site.customize_domain,
+            copyright=site.copyright,
+            privacy_policy=site.privacy_policy,
+            input_placeholder=site.input_placeholder,
+            custom_disclaimer=site.custom_disclaimer,
+            customize_token_strategy=site.customize_token_strategy.value
+            if site.customize_token_strategy is not None
+            else None,
+            prompt_public=site.prompt_public,
+            show_workflow_steps=site.show_workflow_steps,
+            use_icon_as_answer_icon=site.use_icon_as_answer_icon,
+            created_by=site.created_by,
+            created_at=site.created_at,
+            updated_by=site.updated_by,
+            updated_at=site.updated_at,
+        )

api/services/app_preview_details_adapters.py

@@ -0,0 +1,123 @@
+"""Enrich detached app and workflow previews through legacy tool and key providers.
+
+The injected queries close their database sessions before this adapter calls tool,
+plugin or credential providers. TODO: Retire the remaining tool bridge when its
+providers have neutral boundaries. Legacy ToolManager still owns its internal
+global-session and credential-refresh behavior; this adapter does not migrate it.
+"""
+
+from collections.abc import Sequence
+from dataclasses import replace
+from typing import cast, override
+
+from core.agent.tool_configuration import mask_agent_tool_parameters
+from core.plugin.plugin_service import PluginService
+from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.tool_manager import ToolManager
+from core.workflow.environment_variables import load_environment_variables
+from core.workflow.llm_environment_variable import dump_environment_variable
+from models.model import AppMode
+from models.provider_ids import GenericProviderID
+from services.app_preview_details_service import (
+    AppPreviewDeletedTool,
+    AppPreviewDetail,
+    AppPreviewDetails,
+    AppPreviewDetailsQuery,
+    AppPreviewObject,
+    AppPreviewWorkflow,
+)
+from services.app_preview_query_service import AppPreviewRef
+
+
+class AppPreviewDetailsRuntime(AppPreviewDetails):
+    def __init__(self, *, details: AppPreviewDetailsQuery) -> None:
+        self._details: AppPreviewDetailsQuery = details
+
+    @override
+    def get_detail(self, *, app: AppPreviewRef, account_id: str, active_workspace_id: str) -> AppPreviewDetail:
+        record = self._details.get_detail(app=app, account_id=account_id)
+        detail = record.detail
+        configuration = detail.model_config
+        agent_mode = cast(AppPreviewObject, configuration["agent_mode"]) if configuration is not None else None
+        tools = tuple(cast(Sequence[AppPreviewObject], agent_mode.get("tools", []))) if agent_mode else ()
+        # Preserve the response compatibility without the legacy model's App.mode
+        # update and commit. A preview query must leave persisted values unchanged.
+        mode = detail.mode
+        if agent_mode and agent_mode.get("enabled", False) and agent_mode.get("strategy") in {"function_call", "react"}:
+            mode = AppMode.AGENT_CHAT
+        if mode == AppMode.AGENT_CHAT and agent_mode is not None and configuration is not None:
+            configuration = {
+                **configuration,
+                "agent_mode": mask_agent_tool_parameters(
+                    agent_mode=agent_mode,
+                    app_id=app.app_id,
+                    tenant_id=active_workspace_id,
+                    user_id=account_id,
+                ),
+            }
+        return replace(
+            detail,
+            mode=mode,
+            model_config=configuration,
+            deleted_tools=self._get_deleted_tools(
+                tenant_id=app.tenant_id, tools=tools, existing_api_provider_ids=record.existing_api_provider_ids
+            ),
+        )
+
+    @staticmethod
+    def _get_deleted_tools(
+        *, tenant_id: str, tools: Sequence[AppPreviewObject], existing_api_provider_ids: frozenset[str]
+    ) -> tuple[AppPreviewDeletedTool, ...]:
+        builtin_providers: dict[str, GenericProviderID] = {}
+        for tool in tools:
+            if len(tool) < 4 or tool.get("provider_type") != ToolProviderType.BUILT_IN:
+                continue
+            provider_id = tool.get("provider_id")
+            if not isinstance(provider_id, str) or not provider_id:
+                continue
+            try:
+                ToolManager.get_hardcoded_provider(provider_id)
+            except KeyError:
+                try:
+                    builtin_providers[provider_id] = GenericProviderID(provider_id)
+                except ValueError:
+                    continue
+
+        missing_builtin_ids: set[str] = set()
+        if builtin_providers:
+            existence = PluginService.check_tools_existence(tenant_id, list(builtin_providers.values()))
+            missing_builtin_ids = {
+                provider_id for provider_id, exists in zip(builtin_providers, existence, strict=True) if not exists
+            }
+
+        deleted: list[AppPreviewDeletedTool] = []
+        for tool in tools:
+            if len(tool) < 4:
+                continue
+            provider_type = tool.get("provider_type")
+            if provider_type not in (ToolProviderType.API, ToolProviderType.BUILT_IN):
+                continue
+            provider_id = tool.get("provider_id")
+            tool_name = tool.get("tool_name")
+            if not isinstance(provider_id, str) or not provider_id or not isinstance(tool_name, str):
+                continue
+            if (provider_type == ToolProviderType.API and provider_id not in existing_api_provider_ids) or (
+                provider_type == ToolProviderType.BUILT_IN and provider_id in missing_builtin_ids
+            ):
+                deleted.append(
+                    AppPreviewDeletedTool(type=str(provider_type), tool_name=tool_name, provider_id=provider_id)
+                )
+        return tuple(deleted)
+
+    @override
+    def get_workflow(self, *, app: AppPreviewRef) -> AppPreviewWorkflow:
+        record = self._details.get_workflow(app=app)
+        variables = load_environment_variables(
+            tenant_id=record.tenant_id, serialized_variables=record.environment_variables_json
+        )
+        return replace(
+            record.workflow,
+            environment_variables=tuple(
+                cast(AppPreviewObject, dump_environment_variable(variable, mode="json")) for variable in variables
+            ),
+        )

api/services/app_preview_details_service.py

@@ -0,0 +1,141 @@
+"""Detached app and workflow details for an already admitted preview."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Protocol
+
+from pydantic import JsonValue
+
+from services.app_preview_query_service import AppPreviewRef
+
+type AppPreviewModelConfig = Mapping[str, JsonValue | datetime]
+type AppPreviewObject = Mapping[str, JsonValue]
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewAudit:
+    id: str
+    created_by: str | None
+    created_at: datetime | None
+    updated_by: str | None
+    updated_at: datetime | None
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewTag:
+    id: str
+    name: str
+    type: str
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewDeletedTool:
+    type: str
+    tool_name: str
+    provider_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewDetailSite:
+    code: str | None
+    title: str
+    icon_type: str | None
+    icon: str | None
+    icon_background: str | None
+    description: str | None
+    default_language: str
+    chat_color_theme: str | None
+    chat_color_theme_inverted: bool | None
+    customize_domain: str | None
+    copyright: str | None
+    privacy_policy: str | None
+    input_placeholder: str | None
+    custom_disclaimer: str | None
+    customize_token_strategy: str | None
+    prompt_public: bool | None
+    show_workflow_steps: bool | None
+    use_icon_as_answer_icon: bool | None
+    created_by: str | None
+    created_at: datetime | None
+    updated_by: str | None
+    updated_at: datetime | None
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewDetail:
+    id: str
+    name: str
+    description: str | None
+    mode: str
+    icon_type: str | None
+    icon: str | None
+    icon_background: str | None
+    enable_site: bool
+    enable_api: bool
+    model_config: AppPreviewModelConfig | None
+    workflow: AppPreviewAudit | None
+    use_icon_as_answer_icon: bool | None
+    max_active_requests: int | None
+    created_by: str | None
+    created_at: datetime | None
+    updated_by: str | None
+    updated_at: datetime | None
+    deleted_tools: tuple[AppPreviewDeletedTool, ...]
+    tags: tuple[AppPreviewTag, ...]
+    site: AppPreviewDetailSite
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewDetailRecord:
+    """Stored detail values and owner-scoped API providers before tool enrichment."""
+
+    detail: AppPreviewDetail
+    existing_api_provider_ids: frozenset[str]
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewAccount:
+    id: str
+    name: str | None
+    email: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewWorkflow:
+    id: str
+    graph: AppPreviewObject
+    features: AppPreviewObject
+    hash: str | None
+    version: str | None
+    marked_name: str | None
+    marked_comment: str | None
+    created_by: AppPreviewAccount | None
+    created_at: datetime | None
+    updated_by: AppPreviewAccount | None
+    updated_at: datetime | None
+    tool_published: bool
+    environment_variables: tuple[AppPreviewObject, ...]
+    conversation_variables: tuple[AppPreviewObject, ...]
+    rag_pipeline_variables: tuple[AppPreviewObject, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class AppPreviewWorkflowRecord:
+    """Workflow data with environment variables still in their stored, encrypted form."""
+
+    workflow: AppPreviewWorkflow
+    tenant_id: str
+    environment_variables_json: str | None
+
+
+class AppPreviewDetailsQuery(Protocol):
+    def get_detail(self, *, app: AppPreviewRef, account_id: str) -> AppPreviewDetailRecord: ...
+
+    def get_workflow(self, *, app: AppPreviewRef) -> AppPreviewWorkflowRecord: ...
+
+
+class AppPreviewDetails(Protocol):
+    def get_detail(self, *, app: AppPreviewRef, account_id: str, active_workspace_id: str) -> AppPreviewDetail: ...
+
+    def get_workflow(self, *, app: AppPreviewRef) -> AppPreviewWorkflow: ...

api/services/app_service.py

@@ -1,24 +1,22 @@
 import json
 import logging
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Literal, NotRequired, TypedDict, cast, override
 
 import sqlalchemy as sa
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, JsonValue
 from sqlalchemy import ColumnElement, delete, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.model_template import default_app_templates
-from core.agent.entities import AgentToolEntity
 from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
+from core.agent.tool_configuration import mask_agent_tool_parameters
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
-from core.tools.tool_manager import ToolManager
-from core.tools.utils.configuration import ToolParameterConfigurationManager
 from enums import DeploymentEdition
 from events.app_event import app_was_created, app_was_deleted, app_was_updated
 from extensions.ext_database import db  # noqa: F401
@@ -339,13 +337,6 @@ class AppService:
     ) -> App | None:
         return session.get(App, app_id)
 
-    @staticmethod
-    def get_normal_app_by_id(
-        app_id: str,
-        session: Session,
-    ) -> App | None:
-        return session.scalar(select(App).where(App.id == app_id, App.status == "normal").limit(1))
-
     @staticmethod
     def get_visible_app_by_id(
         app_id: str,
@@ -764,42 +755,12 @@ class AppService:
             model_config = app.app_model_config_with_session(session=session)
             if not model_config:
                 return app
-            agent_mode = model_config.agent_mode_dict
-            # decrypt agent tool parameters if it's secret-input
-            for tool in agent_mode.get("tools") or []:
-                if not isinstance(tool, dict) or len(tool.keys()) <= 3:
-                    continue
-                typed_tool = {key: value for key, value in tool.items() if isinstance(key, str)}
-                if len(typed_tool) != len(tool):
-                    continue
-                agent_tool_entity = AgentToolEntity.model_validate(typed_tool)
-                # get tool
-                try:
-                    tool_runtime = ToolManager.get_agent_tool_runtime(
-                        tenant_id=current_user.current_tenant_id,
-                        app_id=app.id,
-                        agent_tool=agent_tool_entity,
-                        user_id=current_user.id,
-                    )
-                    manager = ToolParameterConfigurationManager(
-                        tenant_id=current_user.current_tenant_id,
-                        tool_runtime=tool_runtime,
-                        provider_name=agent_tool_entity.provider_id,
-                        provider_type=agent_tool_entity.provider_type,
-                        identity_id=f"AGENT.{app.id}",
-                    )
-
-                    # get decrypted parameters
-                    if agent_tool_entity.tool_parameters:
-                        parameters = manager.decrypt_tool_parameters(agent_tool_entity.tool_parameters or {})
-                        masked_parameter = manager.mask_tool_parameters(parameters or {})
-                    else:
-                        masked_parameter = {}
-
-                    # override tool parameters
-                    tool["tool_parameters"] = masked_parameter
-                except Exception:
-                    logger.exception("Failed to mask agent tool parameters for tool %s", agent_tool_entity.tool_name)
+            agent_mode = mask_agent_tool_parameters(
+                agent_mode=cast(Mapping[str, JsonValue], model_config.agent_mode_dict),
+                app_id=app.id,
+                tenant_id=current_user.current_tenant_id,
+                user_id=current_user.id,
+            )
 
             # override agent mode
             if model_config:

api/tests/unit_tests/controllers/console/app/test_wraps.py

@@ -1,16 +1,11 @@
 from __future__ import annotations
 
-from contextlib import nullcontext
 from inspect import getsource
-from types import SimpleNamespace
-from unittest.mock import MagicMock
 from uuid import uuid4
 
 import pytest
-from sqlalchemy import event, text
 from sqlalchemy.orm import Session
 
-from controllers.common import session as session_module
 from controllers.common.session import with_session
 from controllers.console.app import completion as completion_module
 from controllers.console.app import workflow as workflow_module
@@ -58,60 +53,6 @@ def test_get_app_model_rejects_wrong_mode(monkeypatch: pytest.MonkeyPatch, sqlit
         handler(app_id=app_model.id)
 
 
-def test_load_previewable_app_model_rejects_app_outside_preview_admission(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    session = MagicMock(spec=Session)
-    app_loader = MagicMock()
-    recommended_app_queries = MagicMock()
-    recommended_app_queries.is_previewable.return_value = False
-    monkeypatch.setattr(
-        wraps_module,
-        "application_services",
-        lambda: SimpleNamespace(recommended_app_queries=recommended_app_queries),
-    )
-    monkeypatch.setattr(wraps_module.AppService, "get_normal_app_by_id", app_loader)
-
-    assert wraps_module._load_previewable_app_model(session, "app-1") is None
-    recommended_app_queries.is_previewable.assert_called_once_with("app-1")
-    app_loader.assert_not_called()
-
-
-def test_load_previewable_app_model_rejects_non_normal_app(
-    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
-) -> None:
-    app_model = _persist_app(sqlite_session)
-    app_id = app_model.id
-    sqlite_session.execute(text("UPDATE apps SET status = 'disabled' WHERE id = :app_id"), {"app_id": app_id})
-    sqlite_session.commit()
-    recommended_app_queries = MagicMock()
-    recommended_app_queries.is_previewable.return_value = True
-    monkeypatch.setattr(
-        wraps_module,
-        "application_services",
-        lambda: SimpleNamespace(recommended_app_queries=recommended_app_queries),
-    )
-
-    assert wraps_module._load_previewable_app_model(sqlite_session, app_id) is None
-
-
-def test_get_previewable_app_model_rejects_app_outside_preview_admission(
-    monkeypatch: pytest.MonkeyPatch, unbound_session: Session
-) -> None:
-    app_loader = MagicMock(return_value=None)
-    monkeypatch.setattr(wraps_module, "_load_previewable_app_model", app_loader)
-
-    class Handler:
-        @wraps_module.get_previewable_app_model
-        def get(self, _injected_session, app_model):
-            return app_model.id
-
-    with pytest.raises(AppNotFoundError):
-        Handler().get(unbound_session, app_id="app-1")
-
-    app_loader.assert_called_once_with(unbound_session, "app-1")
-
-
 def test_get_app_model_requires_app_id() -> None:
     @wraps_module.get_app_model
     def handler(app_model):
@@ -144,60 +85,6 @@ def test_get_app_model_prefers_injected_session(
         assert Handler().get(sqlite_session, app_id=app_model.id) == app_model.id
 
 
-def test_preview_admission_precedes_request_session_transaction(
-    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
-) -> None:
-    app_model = _persist_app(sqlite_session)
-    app_id = app_model.id
-    sqlite_session.rollback()
-    request_transaction_begins = 0
-
-    def record_request_transaction_begin(_session, _transaction, _connection) -> None:
-        nonlocal request_transaction_begins
-        request_transaction_begins += 1
-
-    event.listen(sqlite_session, "after_begin", record_request_transaction_begin)
-    recommended_app_queries = MagicMock()
-
-    def assert_request_session_has_not_started(_app_id: str) -> bool:
-        assert request_transaction_begins == 0
-        assert sqlite_session.in_transaction() is False
-        return True
-
-    recommended_app_queries.is_previewable.side_effect = assert_request_session_has_not_started
-    monkeypatch.setattr(
-        wraps_module,
-        "application_services",
-        lambda: SimpleNamespace(recommended_app_queries=recommended_app_queries),
-    )
-    monkeypatch.setattr(
-        wraps_module.db,
-        "session",
-        SimpleNamespace(scalar=lambda *_args, **_kwargs: pytest.fail("db.session should not be used")),
-    )
-    monkeypatch.setattr(session_module.session_factory, "create_session", lambda: nullcontext(sqlite_session))
-
-    class Handler:
-        @with_session(write=False)
-        @wraps_module.get_previewable_app_model(None)
-        def get(self, injected_session, app_model):
-            assert injected_session is sqlite_session
-            return app_model.id
-
-    assert Handler().get(app_id=app_id) == app_id
-    recommended_app_queries.is_previewable.assert_called_once_with(app_id)
-    assert request_transaction_begins == 1
-
-
-def test_get_previewable_app_model_requires_injected_session() -> None:
-    @wraps_module.get_previewable_app_model(None)
-    def handler(app_model):
-        return app_model.id
-
-    with pytest.raises(RuntimeError, match="requires @with_session"):
-        handler(app_id="app-1")
-
-
 @pytest.mark.parametrize(
     "resource",
     [

api/tests/unit_tests/controllers/console/explore/test_trial.py

@@ -1,5 +1,3 @@
-from datetime import UTC, datetime
-from inspect import getsource, signature
 from inspect import unwrap as inspect_unwrap
 from io import BytesIO
 from types import SimpleNamespace
@@ -8,7 +6,6 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 from flask import Flask, request
-from sqlalchemy.orm import Session
 
 import controllers.console.explore.trial as module
 from controllers.console.app.error import (
@@ -24,14 +21,9 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.helper import encrypter
-from core.workflow.llm_environment_variable import LLMEnvironmentVariable
 from graphon.model_runtime.errors.invoke import InvokeError
-from graphon.variables import SecretVariable, StringVariable
 from models import Account
 from models.model import App, AppMode
-from models.tools import WorkflowToolProvider
-from models.workflow import Workflow
 from services.app_ref_service import AppRef, MessageRef
 from services.errors.audio import SpeechToTextDisabledServiceError
 from tests.unit_tests.model_factories import make_app
@@ -77,39 +69,6 @@ def test_trial_workflow_uses_trial_scoped_simple_account_model() -> None:
     assert module.simple_account_model.__schema__["properties"].keys() >= {"id", "name", "email"}
 
 
-@pytest.mark.parametrize(
-    "api_type",
-    [module.AppApi, module.AppWorkflowApi],
-)
-def test_preview_handlers_use_explicit_read_session(api_type: type) -> None:
-    source = getsource(api_type.get)
-
-    assert "@with_session(write=False)\n    @get_previewable_app_model(None)" in source
-    assert tuple(signature(api_type.get).parameters)[:3] == ("self", "session", "app_model")
-
-
-def test_trial_app_detail_serializes_with_explicit_session(
-    app: Flask, monkeypatch: pytest.MonkeyPatch, unbound_session: Session
-) -> None:
-    app_model = _app(app_id="app-1", mode=AppMode.CHAT)
-    response_view = MagicMock()
-    get_app = MagicMock(return_value=app_model)
-    build_view = MagicMock(return_value=response_view)
-    validated = MagicMock()
-    validated.model_dump.return_value = {"id": "app-1"}
-    monkeypatch.setattr(module, "AppService", lambda: SimpleNamespace(get_app=get_app))
-    monkeypatch.setattr(module, "AppResponseView", build_view)
-    monkeypatch.setattr(module.TrialAppDetailResponse, "model_validate", MagicMock(return_value=validated))
-
-    with app.test_request_context("/"):
-        result = unwrap(module.AppApi.get)(module.AppApi(), unbound_session, app_model)
-
-    assert result == {"id": "app-1"}
-    get_app.assert_called_once_with(app_model, session=unbound_session)
-    build_view.assert_called_once_with(app_model, session=unbound_session)
-    module.TrialAppDetailResponse.model_validate.assert_called_once_with(response_view, from_attributes=True)
-
-
 class TestTrialChatAudioApi:
     def test_success(
         self,
@@ -532,109 +491,6 @@ class TestTrialChatTextApi:
                 )
 
 
-class TestAppWorkflowApi:
-    def test_uses_injected_session(self, sqlite_session: Session) -> None:
-        api = module.AppWorkflowApi()
-        method = unwrap(api.get)
-        created_by = Account(name="Creator", email="creator@example.com")
-        created_by.id = "account-1"
-        app_model = _app(app_id="app-1", mode=AppMode.WORKFLOW)
-        with patch("models.workflow.encrypter.encrypt_token", return_value="encrypted-secret"):
-            workflow = Workflow.new(
-                tenant_id=app_model.tenant_id,
-                app_id=app_model.id,
-                type="workflow",
-                version="draft",
-                graph='{"nodes": []}',
-                features="{}",
-                created_by=created_by.id,
-                environment_variables=[
-                    SecretVariable(
-                        id="env-secret",
-                        name="api_key",
-                        value="plaintext-secret",
-                    ),
-                    LLMEnvironmentVariable(
-                        id="env-llm",
-                        name="shared_model",
-                        value={"provider": "provider", "name": "model", "mode": "chat"},
-                    ),
-                ],
-                conversation_variables=[
-                    StringVariable(
-                        id="conversation-variable-1",
-                        name="topic",
-                        value="sqlite",
-                        selector=["conversation", "topic"],
-                    )
-                ],
-                rag_pipeline_variables=[],
-            )
-        workflow.id = "workflow-1"
-        workflow.created_at = datetime(2024, 1, 1, tzinfo=UTC)
-        workflow.updated_at = datetime(2024, 1, 2, tzinfo=UTC)
-        app_model.workflow_id = workflow.id
-        tool_provider = WorkflowToolProvider(
-            name="trial-workflow",
-            label="Trial Workflow",
-            icon="icon",
-            app_id=app_model.id,
-            version="1.0.0",
-            user_id=created_by.id,
-            tenant_id=app_model.tenant_id,
-            description="Trial workflow provider",
-            parameter_configuration="[]",
-        )
-        sqlite_session.add_all([created_by, app_model, workflow, tool_provider])
-        sqlite_session.commit()
-
-        with patch("models.workflow.encrypter.decrypt_token", return_value="plaintext-secret"):
-            result = method(api, sqlite_session, app_model)
-
-        assert result == {
-            "id": "workflow-1",
-            "graph": {"nodes": []},
-            "features": {},
-            "hash": workflow.unique_hash,
-            "version": "draft",
-            "marked_name": "",
-            "marked_comment": "",
-            "created_by": {"id": "account-1", "name": "Creator", "email": "creator@example.com"},
-            "created_at": 1704067200,
-            "updated_by": None,
-            "updated_at": 1704153600,
-            "tool_published": True,
-            "environment_variables": [
-                {
-                    "value_type": "secret",
-                    "value": encrypter.full_mask_token(),
-                    "id": "env-secret",
-                    "name": "api_key",
-                    "description": "",
-                    "selector": ["env", "api_key"],
-                },
-                {
-                    "value_type": "llm",
-                    "value": {"provider": "provider", "name": "model", "mode": "chat"},
-                    "id": "env-llm",
-                    "name": "shared_model",
-                    "description": "",
-                    "selector": ["env", "shared_model"],
-                },
-            ],
-            "conversation_variables": [
-                {
-                    "id": "conversation-variable-1",
-                    "name": "topic",
-                    "value_type": "string",
-                    "value": "sqlite",
-                    "description": "",
-                }
-            ],
-            "rag_pipeline_variables": [],
-        }
-
-
 class TestTrialChatAudioApiExceptionHandlers:
     def test_provider_not_init(self, app: Flask, trial_app_chat: App, account: Account) -> None:
         api = module.TrialChatAudioApi()

api/tests/unit_tests/controllers/console/explore/test_trial_preview_details.py

@@ -0,0 +1,803 @@
+"""Preview detail contracts through real HTTP, services, and SQLite queries."""
+
+import json
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime
+from types import SimpleNamespace
+from typing import Literal, cast, override
+from uuid import uuid4
+
+import pytest
+from flask import Flask, got_request_exception
+from pydantic import JsonValue
+from sqlalchemy import Connection, Engine, delete, event, select, text, update
+from sqlalchemy.engine import ExecutionContext
+from sqlalchemy.orm import Session, SessionTransaction, sessionmaker
+from sqlalchemy.pool import QueuePool
+from werkzeug.test import TestResponse
+
+import controllers.console.explore.trial as trial_module
+import controllers.console.wraps as console_wraps
+import libs.login as login_module
+import services.app_service as app_module
+from constants import COOKIE_NAME_CSRF_TOKEN, HEADER_NAME_CSRF_TOKEN
+from controllers.console import api as console_api
+from controllers.console.app import preview_admission as admission_module
+from core.agent.entities import AgentToolEntity
+from core.helper import encrypter
+from core.plugin.plugin_service import PluginService
+from core.tools.__base.tool import Tool
+from core.tools.entities.tool_entities import ToolParameter
+from core.tools.tool_manager import ToolManager
+from core.workflow.llm_environment_variable import LLMEnvironmentVariable
+from enums import DeploymentEdition
+from extensions.ext_login import DifyLoginManager, unauthorized_handler
+from graphon.variables import SecretVariable, StringVariable
+from libs.external_api import ExternalApi
+from libs.token import generate_csrf_token
+from machinery.errors import ActiveWorkspaceRequiredError
+from models import Account, AccountTrialAppRecord, App, AppMode, Tenant, TrialApp
+from models.account import AccountStatus
+from models.enums import CustomizeTokenStrategy, TagType
+from models.model import AppModelConfig, IconType, Site, Tag, TagBinding
+from models.provider_ids import GenericProviderID
+from models.tools import WorkflowToolProvider
+from models.workflow import Workflow
+from repositories.app_preview_query_repository import AppPreviewQueryRepository
+from repositories.trial_app_repository import TrialAppRepository
+from services.app_preview_details_adapters import AppPreviewDetailsRuntime
+from services.app_preview_query_service import AppPreviewQueryService
+from services.app_service import AppResponseView, AppService
+from services.recommended_app_query_service import (
+    RecommendedAppCatalogPage,
+    RecommendedAppDetailRecord,
+    RecommendedAppQueryService,
+)
+
+_Endpoint = Literal["detail", "workflows"]
+_CREATED_AT = datetime(2024, 1, 1)
+_UPDATED_AT = datetime(2024, 1, 2)
+
+
+@dataclass
+class _Catalog:
+    engine: Engine
+    ids: set[str] = field(default_factory=set)
+    calls: list[str] = field(default_factory=list)
+
+    def contains(self, app_id: str) -> bool:
+        assert isinstance(self.engine.pool, QueuePool)
+        assert self.engine.pool.checkedout() == 0
+        self.calls.append(app_id)
+        return app_id in self.ids
+
+    def list_recommended(self, language: str) -> RecommendedAppCatalogPage:
+        raise AssertionError(f"Unexpected listing for {language}")
+
+    def list_learn_dify(self, language: str) -> RecommendedAppCatalogPage:
+        raise AssertionError(f"Unexpected listing for {language}")
+
+    def get_detail(self, app_id: str) -> RecommendedAppDetailRecord | None:
+        raise AssertionError(f"Unexpected catalog detail for {app_id}")
+
+
+@dataclass
+class _ExternalIO:
+    engine: Engine
+    owner_id: str
+    viewer_workspace_id: str
+    decryptions: list[tuple[str, str]] = field(default_factory=list)
+    tool_calls: list[tuple[str, str, str]] = field(default_factory=list)
+
+    def encrypt(self, tenant_id: str, token: str) -> str:
+        assert tenant_id == self.owner_id
+        assert token == "workflow-secret"
+        return "cipher-workflow"
+
+    def decrypt(self, tenant_id: str, token: str) -> str:
+        assert isinstance(self.engine.pool, QueuePool)
+        assert self.engine.pool.checkedout() == 0
+        self.decryptions.append((tenant_id, token))
+        if token == "cipher-workflow":
+            assert tenant_id == self.owner_id
+            return "workflow-secret"
+        assert token == "cipher-tool"
+        assert tenant_id == self.viewer_workspace_id
+        return "tool-secret"
+
+    def tool_runtime(self, *, tenant_id: str, app_id: str, agent_tool: AgentToolEntity, user_id: str) -> Tool:
+        assert isinstance(self.engine.pool, QueuePool)
+        assert self.engine.pool.checkedout() == 0
+        assert tenant_id == self.viewer_workspace_id
+        assert agent_tool.tool_name == "search"
+        self.tool_calls.append((tenant_id, app_id, user_id))
+        parameter = ToolParameter.model_validate(
+            {"name": "api_key", "label": {"en_US": "Key"}, "type": "secret-input", "form": "form"}
+        )
+        return cast(
+            Tool, _ToolRuntime(SimpleNamespace(parameters=[parameter], identity=SimpleNamespace(name="search")))
+        )
+
+
+@dataclass
+class _ToolRuntime:
+    entity: SimpleNamespace
+
+    def get_runtime_parameters(self) -> list[ToolParameter]:
+        return []
+
+
+@dataclass(frozen=True)
+class _ApplicationServices:
+    app_previews: AppPreviewQueryService
+    app_preview_details: AppPreviewDetailsRuntime
+    recommended_app_queries: RecommendedAppQueryService
+
+
+@dataclass(frozen=True)
+class _Harness:
+    app: Flask
+    owner: Tenant
+    viewer_workspace: Tenant
+    viewer: Account
+    author: Account
+    updater: Account
+    target: App
+    config: AppModelConfig
+    site: Site
+    workflow: Workflow
+    tag: Tag
+    factory: sessionmaker[Session]
+    engine: Engine
+    catalog: _Catalog
+    io: _ExternalIO
+    sessions: list[Session]
+
+    def get(
+        self,
+        endpoint: _Endpoint = "detail",
+        *,
+        method: str = "GET",
+        csrf_token: str | None = None,
+    ) -> TestResponse:
+        suffix = "" if endpoint == "detail" else "/workflows"
+        statements: list[str] = []
+        token = generate_csrf_token(self.viewer.id) if csrf_token is None else csrf_token
+        client = self.app.test_client()
+        client.set_cookie(COOKIE_NAME_CSRF_TOKEN, token)
+
+        def record_statement(
+            _connection: Connection,
+            _cursor: object,
+            statement: str,
+            _parameters: object,
+            _context: ExecutionContext,
+            _executemany: bool,
+        ) -> None:
+            statements.append(statement)
+
+        event.listen(self.engine, "before_cursor_execute", record_statement)
+        try:
+            response = client.open(
+                f"/trial-apps/{self.target.id}{suffix}", method=method, headers={HEADER_NAME_CSRF_TOKEN: token}
+            )
+        finally:
+            event.remove(self.engine, "before_cursor_execute", record_statement)
+        assert not [sql for sql in statements if sql.lstrip().upper().startswith(("INSERT ", "UPDATE ", "DELETE "))]
+        assert response.headers["Content-Type"] == "application/json"
+        if method != "HEAD":
+            assert int(response.headers["Content-Length"]) == len(response.data)
+        assert isinstance(self.engine.pool, QueuePool)
+        assert self.engine.pool.checkedout() == 0
+        assert all(not session.in_transaction() and not session.identity_map for session in self.sessions)
+        return response
+
+    def assert_app_not_queried(self) -> None:
+        assert self.sessions == []
+        assert self.catalog.calls == []
+        assert self.io.tool_calls == []
+        assert self.io.decryptions == []
+
+    def legacy_detail(self) -> dict[str, object]:
+        """The unchanged ORM response view is the pre-migration serialization oracle."""
+        with self.app.test_request_context(f"/trial-apps/{self.target.id}"), self.factory() as session:
+            app = session.get(App, self.target.id)
+            assert app is not None
+            app = AppService().get_app(app, session=session)
+            return trial_module.TrialAppDetailResponse.model_validate(
+                AppResponseView(app, session=session), from_attributes=True
+            ).model_dump(mode="json")
+
+
+@pytest.fixture
+def harness(
+    monkeypatch: pytest.MonkeyPatch,
+    config_overrides: Callable[..., None],
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+) -> _Harness:
+    config_overrides(
+        LOGIN_DISABLED=False,
+        DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY,
+        INIT_PASSWORD="",
+        SECRET_KEY="preview-admission-test-secret-key",
+        CONSOLE_WEB_URL="",
+        CONSOLE_API_URL="",
+        SERVICE_API_URL="",
+        APP_WEB_URL="",
+    )
+    owner = Tenant(name="App owner")
+    viewer_workspace = Tenant(name="Viewer workspace")
+    viewer = Account(name="Viewer", email="viewer@example.com", status=AccountStatus.ACTIVE)
+    viewer._current_tenant = viewer_workspace
+    author = Account(name="Author", email="author@example.com")
+    updater = Account(name="Updater", email="updater@example.com")
+    target = App(
+        id=str(uuid4()),
+        tenant_id=owner.id,
+        name="Preview app",
+        description="",
+        mode=AppMode.CHAT,
+        icon_type=IconType.EMOJI,
+        icon="",
+        icon_background=None,
+        enable_site=False,
+        enable_api=False,
+        max_active_requests=0,
+        created_at=_CREATED_AT,
+        updated_at=_UPDATED_AT,
+    )
+    config = AppModelConfig(
+        app_id=target.id,
+        opening_statement="",
+        suggested_questions="[]",
+        model='{"provider":"example/model","name":"model","mode":"chat","completion_params":{"temperature":0}}',
+        agent_mode='{"enabled":false,"strategy":"function_call","tools":[]}',
+        user_input_form='[{"number":{"label":"Count","variable":"count","default":0,"required":false}}]',
+    )
+    config.created_at = _CREATED_AT
+    config.updated_at = _UPDATED_AT
+    site = Site(
+        app_id=target.id,
+        title="Preview site",
+        default_language="en-US",
+        customize_token_strategy=CustomizeTokenStrategy.UUID,
+        code="preview-code",
+        icon_type=IconType.EMOJI,
+        icon="",
+        icon_background=None,
+        chat_color_theme="",
+        description=None,
+        custom_disclaimer="",
+        show_workflow_steps=False,
+    )
+    site.created_at = _CREATED_AT
+    site.updated_at = _UPDATED_AT
+    io = _ExternalIO(sqlite_engine, owner.id, viewer_workspace.id)
+    monkeypatch.setattr(encrypter, "encrypt_token", io.encrypt)
+    monkeypatch.setattr(encrypter, "decrypt_token", io.decrypt)
+    monkeypatch.setattr(ToolManager, "get_agent_tool_runtime", io.tool_runtime)
+    monkeypatch.setattr(ToolManager, "get_hardcoded_provider", lambda _provider_id: object())
+    with sqlite_session_factory.begin() as session:
+        session.add_all([owner, viewer_workspace, viewer, author, updater, target, config, site])
+        session.flush()
+        target.created_by = author.id
+        target.updated_by = updater.id
+        target.app_model_config_id = config.id
+        workflow = Workflow.new(
+            tenant_id=owner.id,
+            app_id=target.id,
+            type="workflow",
+            version="draft",
+            graph='{"nodes":[],"edges":[]}',
+            features='{"suggested_questions_after_answer":{"enabled":false}}',
+            created_by=author.id,
+            environment_variables=[
+                SecretVariable(id="secret", name="api_key", value="workflow-secret"),
+                LLMEnvironmentVariable(
+                    id="llm", name="shared_model", value={"provider": "provider", "name": "model", "mode": "chat"}
+                ),
+            ],
+            conversation_variables=[
+                StringVariable(id="topic", name="topic", value="sqlite", selector=["conversation", "topic"])
+            ],
+            rag_pipeline_variables=[],
+        )
+        workflow.created_at = _CREATED_AT
+        workflow.updated_at = _UPDATED_AT
+        workflow.updated_by = updater.id
+        target.workflow_id = workflow.id
+        tag = Tag(tenant_id=owner.id, type=TagType.APP, name="Owner tag", created_by=author.id)
+        session.add_all([workflow, tag])
+        session.flush()
+        session.add_all(
+            [
+                TrialApp(app_id=target.id, tenant_id=str(uuid4()), trial_limit=0),
+                AccountTrialAppRecord(app_id=target.id, account_id=viewer.id, count=100),
+                TagBinding(tenant_id=owner.id, tag_id=tag.id, target_id=target.id, created_by=author.id),
+                WorkflowToolProvider(
+                    name="trial-workflow",
+                    label="Trial workflow",
+                    icon="icon",
+                    app_id=target.id,
+                    version="1.0.0",
+                    user_id=author.id,
+                    tenant_id=owner.id,
+                    description="Workflow provider",
+                    parameter_configuration="[]",
+                ),
+            ]
+        )
+    factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    sessions: list[Session] = []
+
+    @event.listens_for(factory, "after_begin")
+    def track_session(session: Session, _transaction: SessionTransaction, _connection: Connection) -> None:
+        assert all(previous is session or not previous.in_transaction() for previous in sessions)
+        sessions.append(session)
+
+    catalog = _Catalog(sqlite_engine)
+    recommendations = RecommendedAppQueryService(
+        catalog=catalog, trial_apps=TrialAppRepository(factory), trial_enabled=False
+    )
+    services = _ApplicationServices(
+        app_previews=AppPreviewQueryService(
+            apps=AppPreviewQueryRepository(session_factory=factory), is_previewable=recommendations.is_previewable
+        ),
+        app_preview_details=AppPreviewDetailsRuntime(details=AppPreviewQueryRepository(session_factory=factory)),
+        recommended_app_queries=recommendations,
+    )
+    for module in (trial_module, admission_module):
+        monkeypatch.setattr(module, "application_services", lambda: services)
+    monkeypatch.setattr(login_module, "current_user", viewer)
+    monkeypatch.setattr(app_module, "current_user", viewer)
+    monkeypatch.setattr(console_wraps, "_is_setup_completed", lambda: True)
+    app = Flask(__name__)
+    app.config.update(TESTING=True, RESTX_ERROR_404_HELP=False)
+    login_manager = DifyLoginManager()
+    login_manager.init_app(app)
+    login_manager.unauthorized_handler(unauthorized_handler)
+    api = ExternalApi(app)
+    api.error_handlers = console_api.error_handlers.copy()
+    api.add_resource(trial_module.AppApi, "/trial-apps/<uuid:app_id>")
+    api.add_resource(trial_module.AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows")
+    return _Harness(
+        app,
+        owner,
+        viewer_workspace,
+        viewer,
+        author,
+        updater,
+        target,
+        config,
+        site,
+        workflow,
+        tag,
+        sqlite_session_factory,
+        sqlite_engine,
+        catalog,
+        io,
+        sessions,
+    )
+
+
+def _assert_error(response: TestResponse, status: int, code: str) -> None:
+    assert response.status_code == status, response.get_json()
+    body = response.get_json()
+    assert body["status"] == status
+    assert body["code"] == code
+    assert body["message"]
+
+
+@pytest.mark.parametrize(
+    "mode", [AppMode.CHAT, AppMode.COMPLETION, AppMode.ADVANCED_CHAT, AppMode.WORKFLOW, AppMode.AGENT_CHAT]
+)
+def test_app_detail_keeps_original_response_fields_and_stored_audit_times(harness: _Harness, mode: AppMode) -> None:
+    with harness.factory.begin() as session:
+        session.execute(update(App).where(App.id == harness.target.id).values(mode=mode))
+    expected = harness.legacy_detail()
+    if mode == AppMode.AGENT_CHAT:
+        expected_config = expected["model_config"]
+        assert isinstance(expected_config, dict)
+        # Legacy masking could autoflush a response-only config mutation and bump
+        # this timestamp. A read-only preview returns the actual stored audit value.
+        expected_config["updated_at"] = int(_UPDATED_AT.timestamp())
+
+    response = harness.get()
+
+    assert response.status_code == 200, response.get_json()
+    assert response.get_json() == expected
+    body = response.get_json()
+    assert body["max_active_requests"] == 0
+    assert body["enable_site"] is False
+    assert body["tags"] == [{"id": harness.tag.id, "name": "Owner tag", "type": "app"}]
+    assert body["api_base_url"] == "http://localhost/v1"
+    assert body["site"]["app_base_url"] == "http://localhost"
+    assert body["site"]["access_token"] == "preview-code"
+    with harness.factory() as session:
+        assert session.scalar(select(AccountTrialAppRecord.count)) == 100
+        assert (
+            session.scalar(select(AppModelConfig.updated_at).where(AppModelConfig.id == harness.config.id))
+            == _UPDATED_AT
+        )
+
+
+def test_legacy_agent_preview_masks_tools_without_upgrading_mode_or_mutating_config(harness: _Harness) -> None:
+    agent_mode = {
+        "enabled": True,
+        "strategy": "function_call",
+        "tools": [
+            {
+                "provider_type": "builtin",
+                "provider_id": "example",
+                "tool_name": "search",
+                "tool_parameters": {"api_key": "cipher-tool"},
+            }
+        ],
+    }
+    with harness.factory.begin() as session:
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(agent_mode=json.dumps(agent_mode))
+        )
+        app_updated_at = session.scalar(select(App.updated_at).where(App.id == harness.target.id))
+        config_updated_at = session.scalar(
+            select(AppModelConfig.updated_at).where(AppModelConfig.id == harness.config.id)
+        )
+
+    response = harness.get()
+
+    assert response.status_code == 200, response.get_json()
+    body = response.get_json()
+    assert body["mode"] == "agent-chat"
+    assert body["model_config"]["agent_mode"]["tools"][0]["tool_parameters"] == {"api_key": "to*******et"}
+    assert harness.io.tool_calls == [(harness.viewer_workspace.id, harness.target.id, harness.viewer.id)]
+    assert harness.io.decryptions == [(harness.viewer_workspace.id, "cipher-tool")]
+    assert harness.viewer_workspace.id != harness.owner.id
+    with harness.factory() as session:
+        app = session.get(App, harness.target.id)
+        config = session.get(AppModelConfig, harness.config.id)
+        assert app is not None
+        assert config is not None
+        assert app.mode == AppMode.CHAT
+        assert app.updated_at == app_updated_at
+        assert config.agent_mode == json.dumps(agent_mode)
+        assert config.updated_at == config_updated_at
+
+
+@pytest.mark.parametrize(
+    "provider_fields",
+    [{}, {"provider_id": None}, {"provider_id": 42}, {"provider_id": ""}, {"provider_id": "invalid/provider"}],
+    ids=["missing", "null", "non-string", "empty", "invalid-format"],
+)
+def test_invalid_builtin_config_does_not_break_detail_or_valid_tool_masking(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, provider_fields: dict[str, JsonValue]
+) -> None:
+    agent_mode: dict[str, JsonValue] = {
+        "enabled": True,
+        "strategy": "function_call",
+        "tools": [
+            {
+                "provider_type": "builtin",
+                "tool_name": "broken",
+                "tool_parameters": {"api_key": "cipher-secret"},
+                "credential_id": None,
+                **provider_fields,
+            },
+            {
+                "provider_type": "builtin",
+                "provider_id": "example",
+                "tool_name": "search",
+                "tool_parameters": {"api_key": "cipher-tool"},
+            },
+        ],
+    }
+    with harness.factory.begin() as session:
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(agent_mode=json.dumps(agent_mode))
+        )
+
+    def hardcoded_provider(provider_id: str) -> object:
+        if provider_id == "example":
+            return object()
+        raise KeyError(provider_id)
+
+    def tool_runtime(*, tenant_id: str, app_id: str, agent_tool: AgentToolEntity, user_id: str) -> Tool:
+        if agent_tool.provider_id != "example":
+            raise ValueError("Invalid historical provider ID")
+        return harness.io.tool_runtime(tenant_id=tenant_id, app_id=app_id, agent_tool=agent_tool, user_id=user_id)
+
+    plugin_queries: list[str] = []
+
+    def check_plugins(tenant_id: str, provider_ids: Sequence[GenericProviderID]) -> Sequence[bool]:
+        assert tenant_id == harness.owner.id
+        plugin_queries.extend(str(provider_id) for provider_id in provider_ids)
+        return [False] * len(provider_ids)
+
+    monkeypatch.setattr(ToolManager, "get_hardcoded_provider", hardcoded_provider)
+    monkeypatch.setattr(ToolManager, "get_agent_tool_runtime", tool_runtime)
+    monkeypatch.setattr(PluginService, "check_tools_existence", check_plugins)
+
+    response = harness.get()
+
+    assert response.status_code == 200, response.get_json()
+    body = response.get_json()
+    assert body["deleted_tools"] == []
+    assert body["model_config"]["agent_mode"]["tools"][0]["tool_parameters"] == {}
+    assert body["model_config"]["agent_mode"]["tools"][1]["tool_parameters"] == {"api_key": "to*******et"}
+    assert b"cipher-secret" not in response.data
+    assert plugin_queries == []
+    assert harness.io.tool_calls == [(harness.viewer_workspace.id, harness.target.id, harness.viewer.id)]
+    with harness.factory() as session:
+        assert session.scalar(
+            select(AppModelConfig.agent_mode).where(AppModelConfig.id == harness.config.id)
+        ) == json.dumps(agent_mode)
+
+
+@pytest.mark.parametrize("failure", ["short-config", "runtime", "masking"])
+def test_unmaskable_tool_parameters_are_removed_from_detail_without_changing_storage(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, failure: str
+) -> None:
+    broken_tool: dict[str, JsonValue] = {
+        "provider_type": "builtin",
+        "tool_parameters": {"api_key": "cipher-secret"},
+    }
+    if failure != "short-config":
+        broken_tool.update(provider_id="broken", tool_name="broken")
+    agent_mode: dict[str, JsonValue] = {
+        "enabled": True,
+        "strategy": "function_call",
+        "tools": [
+            broken_tool,
+            {
+                "provider_type": "builtin",
+                "provider_id": "example",
+                "tool_name": "search",
+                "tool_parameters": {"api_key": "cipher-tool"},
+            },
+        ],
+    }
+    with harness.factory.begin() as session:
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(agent_mode=json.dumps(agent_mode))
+        )
+
+    class MaskingFailureRuntime(_ToolRuntime):
+        parameter_queries: int = 0
+
+        @override
+        def get_runtime_parameters(self) -> list[ToolParameter]:
+            self.parameter_queries += 1
+            if self.parameter_queries == 2:
+                raise RuntimeError("Tool metadata became unavailable during masking")
+            return []
+
+    parameter = ToolParameter.model_validate(
+        {"name": "api_key", "label": {"en_US": "Key"}, "type": "secret-input", "form": "form"}
+    )
+    broken_runtime = MaskingFailureRuntime(
+        SimpleNamespace(parameters=[parameter], identity=SimpleNamespace(name="broken"))
+    )
+
+    def tool_runtime(*, tenant_id: str, app_id: str, agent_tool: AgentToolEntity, user_id: str) -> Tool:
+        if agent_tool.provider_id == "example":
+            return harness.io.tool_runtime(tenant_id=tenant_id, app_id=app_id, agent_tool=agent_tool, user_id=user_id)
+        if failure == "runtime":
+            raise RuntimeError("Tool provider is unavailable")
+        return cast(Tool, broken_runtime)
+
+    def decrypt(tenant_id: str, token: str) -> str:
+        if token == "cipher-secret":
+            assert tenant_id == harness.viewer_workspace.id
+            return "plain-secret"
+        return harness.io.decrypt(tenant_id, token)
+
+    monkeypatch.setattr(ToolManager, "get_agent_tool_runtime", tool_runtime)
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+
+    response = harness.get()
+
+    assert response.status_code == 200, response.get_json()
+    tools = response.get_json()["model_config"]["agent_mode"]["tools"]
+    assert tools[0]["tool_parameters"] == {}
+    assert tools[1]["tool_parameters"] == {"api_key": "to*******et"}
+    assert b"cipher-secret" not in response.data
+    assert b"plain-secret" not in response.data
+    assert broken_runtime.parameter_queries == (2 if failure == "masking" else 0)
+    with harness.factory() as session:
+        assert session.scalar(
+            select(AppModelConfig.agent_mode).where(AppModelConfig.id == harness.config.id)
+        ) == json.dumps(agent_mode)
+
+
+def test_workflow_is_anonymous_and_masks_secrets_after_query_session_closes(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(login_module, "current_user", None)
+    monkeypatch.setattr(console_wraps, "_is_setup_completed", lambda: False)
+    response = harness.get("workflows", csrf_token="")
+    assert response.status_code == 200, response.get_json()
+    assert response.get_json() == {
+        "id": harness.workflow.id,
+        "graph": {"nodes": [], "edges": []},
+        "features": {"suggested_questions_after_answer": {"enabled": False}},
+        "hash": harness.workflow.unique_hash,
+        "version": "draft",
+        "marked_name": "",
+        "marked_comment": "",
+        "created_by": {"id": harness.author.id, "name": "Author", "email": "author@example.com"},
+        "created_at": int(_CREATED_AT.timestamp()),
+        "updated_by": {"id": harness.updater.id, "name": "Updater", "email": "updater@example.com"},
+        "updated_at": int(_UPDATED_AT.timestamp()),
+        "tool_published": True,
+        "environment_variables": [
+            {
+                "value_type": "secret",
+                "value": encrypter.full_mask_token(),
+                "id": "secret",
+                "name": "api_key",
+                "description": "",
+                "selector": ["env", "api_key"],
+            },
+            {
+                "value_type": "llm",
+                "value": {"provider": "provider", "name": "model", "mode": "chat"},
+                "id": "llm",
+                "name": "shared_model",
+                "description": "",
+                "selector": ["env", "shared_model"],
+            },
+        ],
+        "conversation_variables": [
+            {"id": "topic", "name": "topic", "value_type": "string", "value": "sqlite", "description": ""}
+        ],
+        "rag_pipeline_variables": [],
+    }
+    assert harness.io.decryptions == [(harness.owner.id, "cipher-workflow")]
+    assert b"workflow-secret" not in response.data
+    assert b"cipher-workflow" not in response.data
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflows"])
+def test_catalog_only_admission_keeps_preview_available(harness: _Harness, endpoint: _Endpoint) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(TrialApp).where(TrialApp.app_id == harness.target.id))
+    harness.catalog.ids.add(harness.target.id)
+    response = harness.get(endpoint)
+    assert response.status_code == 200, response.get_json()
+    assert harness.catalog.calls == [harness.target.id]
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflows"])
+@pytest.mark.parametrize("unavailable", ["unlisted", "disabled", "missing"])
+def test_unavailable_apps_fail_admission(harness: _Harness, endpoint: _Endpoint, unavailable: str) -> None:
+    with harness.factory.begin() as session:
+        if unavailable == "unlisted":
+            session.execute(delete(TrialApp).where(TrialApp.app_id == harness.target.id))
+        elif unavailable == "disabled":
+            session.execute(text("UPDATE apps SET status='disabled' WHERE id=:app_id"), {"app_id": harness.target.id})
+        else:
+            session.execute(delete(App).where(App.id == harness.target.id))
+    _assert_error(harness.get(endpoint), 404, "app_not_found")
+    assert harness.io.decryptions == []
+
+
+def test_anonymous_app_detail_requires_account(harness: _Harness, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(login_module, "current_user", None)
+    response = harness.get()
+    assert response.status_code == 401
+    assert response.get_json() == {"code": "unauthorized", "message": "Unauthorized."}
+    harness.assert_app_not_queried()
+
+
+@pytest.mark.parametrize(("init_password", "error_code"), [("", "not_setup"), ("setup-password", "not_init_validated")])
+def test_app_detail_requires_setup_before_account_or_app_queries(
+    harness: _Harness,
+    monkeypatch: pytest.MonkeyPatch,
+    config_overrides: Callable[..., None],
+    init_password: str,
+    error_code: str,
+) -> None:
+    config_overrides(INIT_PASSWORD=init_password)
+    monkeypatch.setattr(console_wraps, "_is_setup_completed", lambda: False)
+    monkeypatch.setattr(login_module, "current_user", None)
+    _assert_error(harness.get(csrf_token=""), 401, error_code)
+    harness.assert_app_not_queried()
+
+
+@pytest.mark.parametrize("csrf_token", ["", "invalid-signed-token"], ids=["missing", "invalid"])
+def test_app_detail_rejects_invalid_csrf_before_app_queries(harness: _Harness, csrf_token: str) -> None:
+    response = harness.get(csrf_token=csrf_token)
+    _assert_error(response, 401, "unauthorized")
+    assert response.get_json()["message"] == "CSRF token is missing or invalid."
+    harness.assert_app_not_queried()
+
+
+def test_app_detail_rejects_csrf_token_for_another_account(harness: _Harness) -> None:
+    response = harness.get(csrf_token=generate_csrf_token(harness.author.id))
+    _assert_error(response, 401, "unauthorized")
+    assert response.get_json()["message"] == "CSRF token is missing or invalid."
+    harness.assert_app_not_queried()
+
+
+def test_app_detail_rejects_uninitialized_account_before_app_queries(harness: _Harness) -> None:
+    harness.viewer.status = AccountStatus.UNINITIALIZED
+    _assert_error(harness.get(), 400, "account_not_initialized")
+    harness.assert_app_not_queried()
+
+
+def test_missing_active_workspace_uses_shared_console_error_and_monitoring(harness: _Harness) -> None:
+    harness.viewer._current_tenant = None
+    errors: list[BaseException] = []
+
+    def record_error(_sender: object, *, exception: BaseException) -> None:
+        errors.append(exception)
+
+    with got_request_exception.connected_to(record_error):
+        response = harness.get()
+
+    _assert_error(response, 500, "active_workspace_required")
+    assert response.get_json()["message"] == "Internal Server Error"
+    assert len(errors) == 1
+    assert isinstance(errors[0], ActiveWorkspaceRequiredError)
+    harness.assert_app_not_queried()
+
+
+def test_deleted_admitted_account_is_unauthorized(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(Account).where(Account.id == harness.viewer.id))
+    _assert_error(harness.get(), 401, "unauthorized")
+
+
+def test_missing_site_is_specific_error(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(Site).where(Site.app_id == harness.target.id))
+    _assert_error(harness.get(), 403, "app_site_unavailable")
+
+
+def test_missing_model_config_retains_nullable_detail(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(AppModelConfig).where(AppModelConfig.id == harness.config.id))
+    response = harness.get()
+    assert response.status_code == 200, response.get_json()
+    assert response.get_json()["model_config"] is None
+
+
+@pytest.mark.parametrize("missing", ["reference", "workflow"])
+def test_missing_workflow_is_app_unavailable(harness: _Harness, missing: str) -> None:
+    with harness.factory.begin() as session:
+        if missing == "reference":
+            session.execute(update(App).where(App.id == harness.target.id).values({App.workflow_id: None}))
+        else:
+            session.execute(delete(Workflow).where(Workflow.id == harness.workflow.id))
+    _assert_error(harness.get("workflows"), 400, "app_unavailable")
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflows"])
+def test_head_keeps_get_status_and_headers_without_response_body(harness: _Harness, endpoint: _Endpoint) -> None:
+    response = harness.get(endpoint)
+    head = harness.get(endpoint, method="HEAD")
+    assert response.status_code == head.status_code == 200
+    assert head.data == b""
+    assert head.headers == response.headers
+
+
+@pytest.mark.parametrize(
+    ("endpoint", "relation"), [("detail", "config"), ("detail", "workflow"), ("workflows", "workflow")]
+)
+def test_cross_owner_configuration_is_not_exposed(harness: _Harness, endpoint: _Endpoint, relation: str) -> None:
+    with harness.factory.begin() as session:
+        if relation == "config":
+            session.execute(
+                update(AppModelConfig).where(AppModelConfig.id == harness.config.id).values(app_id=str(uuid4()))
+            )
+        else:
+            session.execute(update(Workflow).where(Workflow.id == harness.workflow.id).values(tenant_id=str(uuid4())))
+    _assert_error(harness.get(endpoint), 400, "app_unavailable")
+    assert harness.io.decryptions == []

api/tests/unit_tests/core/workflow/test_environment_variables.py

@@ -0,0 +1,84 @@
+import json
+
+import pytest
+
+from core.helper import encrypter
+from core.workflow.environment_variables import load_environment_variables
+from core.workflow.llm_environment_variable import LLMEnvironmentVariable, dump_environment_variable
+from graphon.variables import FloatVariable, IntegerVariable, SecretVariable, StringVariable
+from graphon.variables.exc import VariableError
+
+
+@pytest.mark.parametrize(
+    ("tenant_id", "serialized_variables"),
+    [(None, "invalid JSON"), ("", "invalid JSON"), ("tenant", None), ("tenant", ""), ("tenant", "{}")],
+)
+def test_load_empty_or_ownerless_environment_variables(tenant_id: str | None, serialized_variables: str | None) -> None:
+    assert load_environment_variables(tenant_id=tenant_id, serialized_variables=serialized_variables) == []
+
+
+def test_load_environment_variables_preserves_order_types_and_decrypts_only_secrets(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    variables = [
+        IntegerVariable(name="count", value=0, selector=["env", "count"]),
+        SecretVariable(name="token", value="ciphertext", selector=["env", "token"]),
+        StringVariable(name="text", value="", selector=["env", "text"]),
+        FloatVariable(name="ratio", value=1.5, selector=["env", "ratio"]),
+        LLMEnvironmentVariable(
+            name="model",
+            value={"provider": "langgenius/anthropic/anthropic", "name": "claude-sonnet", "mode": "chat"},
+            selector=["env", "model"],
+        ),
+    ]
+    serialized = json.dumps({var.name: dump_environment_variable(var, mode="json") for var in variables})
+    decrypt_calls: list[tuple[str, str]] = []
+
+    def decrypt(*, tenant_id: str, token: str) -> str:
+        decrypt_calls.append((tenant_id, token))
+        return "plaintext"
+
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+    loaded = load_environment_variables(tenant_id="owner-tenant", serialized_variables=serialized)
+
+    assert loaded == [variables[0], variables[1].model_copy(update={"value": "plaintext"}), *variables[2:]]
+    assert [type(var) for var in loaded] == [type(var) for var in variables]
+    assert decrypt_calls == [("owner-tenant", "ciphertext")]
+    assert variables[1].value == "ciphertext"
+
+
+def test_load_environment_variables_propagates_invalid_json() -> None:
+    with pytest.raises(json.JSONDecodeError):
+        load_environment_variables(tenant_id="tenant", serialized_variables="invalid JSON")
+
+
+def test_load_environment_variables_validates_all_variables_before_decryption(monkeypatch: pytest.MonkeyPatch) -> None:
+    decrypt_calls: list[tuple[str, str]] = []
+
+    def decrypt(*, tenant_id: str, token: str) -> str:
+        decrypt_calls.append((tenant_id, token))
+        return "plaintext"
+
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+    serialized = json.dumps(
+        {"secret": {"name": "secret", "value_type": "secret", "value": "ciphertext"}, "invalid": {}}
+    )
+
+    with pytest.raises(VariableError, match="missing name"):
+        load_environment_variables(tenant_id="tenant", serialized_variables=serialized)
+    assert decrypt_calls == []
+
+
+def test_load_environment_variables_propagates_decryption_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    failure = RuntimeError("credential store unavailable")
+
+    def decrypt(*, tenant_id: str, token: str) -> str:
+        assert (tenant_id, token) == ("tenant", "ciphertext")
+        raise failure
+
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+    serialized = json.dumps({"secret": {"name": "secret", "value_type": "secret", "value": "ciphertext"}})
+
+    with pytest.raises(RuntimeError) as caught:
+        load_environment_variables(tenant_id="tenant", serialized_variables=serialized)
+    assert caught.value is failure

api/tests/unit_tests/extensions/test_ext_application_services.py

@@ -19,7 +19,8 @@ from extensions import ext_application_services
 from extensions.ext_redis import RedisClientWrapper
 from machinery.context import RequestContext
 from models.account import Account
-from models.model import AccountTrialAppRecord, App, DifySetup, TrialApp
+from models.enums import CustomizeTokenStrategy
+from models.model import AccountTrialAppRecord, App, DifySetup, Site, TrialApp
 from repositories.account_activation_repository import SQLAlchemyAccountActivationRepository
 from repositories.account_integration_repository import SQLAlchemyAccountIntegrationRepository
 from repositories.account_oauth_repository import (
@@ -765,6 +766,43 @@ def test_app_previews_use_the_configured_catalog_and_app_owner(
             services.app_previews.get_access(app_id=other_id)
 
 
+def test_app_preview_details_use_the_configured_database_without_request_globals(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    services = ext_application_services.build_application_services(
+        database_client=sqlite_session_factory,
+        deployment_edition=DeploymentEdition.COMMUNITY,
+        initialization_password="",
+        redis=MagicMock(spec=RedisClientWrapper),
+    )
+    app_id, owner_id, viewer_workspace_id = str(uuid4()), str(uuid4()), str(uuid4())
+    account = Account(name="Preview viewer", email="preview@example.com")
+    with sqlite_session_factory.begin() as session:
+        session.add_all(
+            [
+                account,
+                App(id=app_id, tenant_id=owner_id, name="Preview", mode="chat", enable_site=True, enable_api=False),
+                Site(
+                    app_id=app_id,
+                    title="Preview site",
+                    default_language="en-US",
+                    customize_token_strategy=CustomizeTokenStrategy.UUID,
+                ),
+            ]
+        )
+
+    detail = services.app_preview_details.get_detail(
+        app=AppPreviewRef(app_id=app_id, tenant_id=owner_id),
+        account_id=account.id,
+        active_workspace_id=viewer_workspace_id,
+    )
+
+    assert detail.id == app_id
+    assert detail.name == "Preview"
+    assert detail.site.title == "Preview site"
+    assert detail.model_config is None
+
+
 def test_build_application_services_adapts_enterprise_webapp_access_mode(
     sqlite_session_factory: sessionmaker[Session],
 ) -> None:

api/tests/unit_tests/repositories/test_app_preview_query_repository.py

@@ -1,21 +1,27 @@
 """Preview reads enforce app ownership without adding execution or dataset ACL policy."""
 
-from dataclasses import replace
+import json
+from dataclasses import asdict, replace
 from datetime import datetime
-from typing import Literal
+from typing import Literal, override
 from uuid import uuid4
 
 import pytest
-from sqlalchemy import Engine, select, text
+from sqlalchemy import Connection, Engine, ExecutionContext, event, select, text
 from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import QueuePool
 
-from models.account import Tenant, TenantStatus
+from core.tools.entities.tool_entities import ApiProviderSchemaType
+from graphon.variables import StringVariable
+from models.account import Account, Tenant, TenantStatus
 from models.dataset import Dataset
 from models.enums import CustomizeTokenStrategy
-from models.model import App, AppMode, IconType, Site
+from models.model import App, AppMode, AppModelConfig, IconType, Site
+from models.tools import ApiToolProvider, WorkflowToolProvider
+from models.workflow import Workflow, WorkflowType
 from repositories.app_preview_query_repository import AppPreviewQueryRepository
 from services.app_definition_query_service import AppSiteConfiguration
+from services.app_preview_details_service import AppPreviewAccount
 from services.app_preview_query_service import AppPreviewDataset, AppPreviewRef
 
 _APP_ID = "11111111-1111-1111-1111-111111111111"
@@ -340,3 +346,277 @@ def test_preview_queries_leave_caller_transaction_open_and_uncommitted(
     with sqlite_session_factory() as session:
         assert session.scalar(select(App.name).where(App.id == _APP_ID)) == "Preview"
     assert sqlite_engine.pool.checkedout() == 0
+
+
+@pytest.fixture
+def detail_config(sqlite_session_factory: sessionmaker[Session], preview_ref: AppPreviewRef) -> AppModelConfig:
+    config = AppModelConfig(
+        app_id=preview_ref.app_id,
+        opening_statement="Stored opening",
+        agent_mode=json.dumps(
+            {
+                "enabled": True,
+                "strategy": "react",
+                "tools": [
+                    {
+                        "provider_type": "builtin",
+                        "provider_id": "acme/search/search",
+                        "tool_name": "search",
+                        "tool_parameters": {"api_key": "encrypted-tool-secret"},
+                    }
+                ],
+            }
+        ),
+    )
+    account = Account(name="Viewer", email="preview@example.com")
+    account.id = _CREATOR_ID
+    with sqlite_session_factory.begin() as session:
+        session.add_all([account, config])
+        session.flush()
+        app = session.get(App, preview_ref.app_id)
+        assert app is not None
+        app.app_model_config_id = config.id
+    return config
+
+
+def test_get_detail_returns_read_only_detached_values_before_mode_and_tool_enrichment(
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+    preview_ref: AppPreviewRef,
+    detail_config: AppModelConfig,
+) -> None:
+    closed: list[Session] = []
+    mutated: list[object] = []
+    writes: list[str] = []
+
+    class TrackedSession(Session):
+        @override
+        def close(self) -> None:
+            mutated.extend(self.new)
+            mutated.extend(self.dirty)
+            mutated.extend(self.deleted)
+            super().close()
+            closed.append(self)
+
+    def record_writes(
+        _connection: Connection,
+        _cursor: object,
+        statement: str,
+        _parameters: object,
+        _context: ExecutionContext,
+        _executemany: bool,
+    ) -> None:
+        if statement.lstrip().split(maxsplit=1)[0].upper() in {"INSERT", "UPDATE", "DELETE"}:
+            writes.append(statement)
+
+    factory: sessionmaker[Session] = sessionmaker(bind=sqlite_engine, class_=TrackedSession, expire_on_commit=False)
+    repository = AppPreviewQueryRepository(session_factory=factory)
+    event.listen(sqlite_engine, "before_cursor_execute", record_writes)
+    try:
+        record = repository.get_detail(app=preview_ref, account_id=_CREATOR_ID)
+    finally:
+        event.remove(sqlite_engine, "before_cursor_execute", record_writes)
+
+    assert isinstance(sqlite_engine.pool, QueuePool)
+    assert sqlite_engine.pool.checkedout() == 0
+    assert closed
+    assert all(not session.in_transaction() and not session.identity_map for session in closed)
+    assert mutated == []
+    assert writes == []
+    assert record.detail.id == preview_ref.app_id
+    assert record.detail.mode == AppMode.CHAT
+    assert record.detail.site.title == "Preview Site"
+    assert record.detail.model_config is not None
+    assert record.detail.model_config["opening_statement"] == "Stored opening"
+    assert record.detail.model_config["agent_mode"] == json.loads(detail_config.agent_mode or "{}")
+    assert record.detail.deleted_tools == ()
+    assert record.existing_api_provider_ids == frozenset()
+
+    def serialize_date(value: object) -> str:
+        assert isinstance(value, datetime), f"Non-data value escaped repository: {type(value)}"
+        return value.isoformat()
+
+    json.dumps(asdict(record.detail), default=serialize_date)
+    with sqlite_session_factory() as session:
+        assert session.scalar(select(App.mode).where(App.id == preview_ref.app_id)) == AppMode.CHAT
+        assert session.scalar(select(AppModelConfig.agent_mode).where(AppModelConfig.id == detail_config.id)) == (
+            detail_config.agent_mode
+        )
+
+
+def test_get_detail_resolves_only_configured_api_providers_owned_by_app_tenant(
+    sqlite_session_factory: sessionmaker[Session], preview_ref: AppPreviewRef, detail_config: AppModelConfig
+) -> None:
+    owner_provider, viewer_provider, unconfigured_provider = [
+        ApiToolProvider(
+            name=name,
+            icon="icon",
+            schema="{}",
+            schema_type_str=ApiProviderSchemaType.OPENAPI,
+            user_id=_CREATOR_ID,
+            tenant_id=tenant_id,
+            description=name,
+            tools_str="[]",
+            credentials_str="{}",
+        )
+        for name, tenant_id in [
+            ("Owner provider", preview_ref.tenant_id),
+            ("Viewer provider", _DECOY_TENANT_ID),
+            ("Unconfigured provider", preview_ref.tenant_id),
+        ]
+    ]
+    with sqlite_session_factory.begin() as session:
+        session.add_all([owner_provider, viewer_provider, unconfigured_provider])
+        config = session.get(AppModelConfig, detail_config.id)
+        assert config is not None
+        config.agent_mode = json.dumps(
+            {
+                "enabled": False,
+                "tools": [
+                    {
+                        "provider_type": "api",
+                        "provider_id": provider_id,
+                        "tool_name": "search",
+                        "tool_parameters": {},
+                    }
+                    for provider_id in [owner_provider.id, viewer_provider.id, owner_provider.id, str(uuid4())]
+                ],
+            }
+        )
+
+    record = AppPreviewQueryRepository(session_factory=sqlite_session_factory).get_detail(
+        app=preview_ref, account_id=_CREATOR_ID
+    )
+
+    assert record.existing_api_provider_ids == frozenset({owner_provider.id})
+    assert record.detail.deleted_tools == ()
+
+
+@pytest.mark.parametrize(
+    "stored_environment",
+    [
+        pytest.param("{invalid-json", id="unparsed-invalid-json"),
+        pytest.param(
+            '{"secret":{"id":"secret","name":"api_key","value_type":"secret",'
+            '"value":"encrypted-workflow-secret","selector":[]}}',
+            id="undecrypted-secret",
+        ),
+    ],
+)
+def test_get_workflow_returns_detached_data_and_raw_environment_after_session_closes(
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+    preview_ref: AppPreviewRef,
+    stored_environment: str,
+) -> None:
+    author = Account(name="Author", email="author@example.com")
+    author.id = _CREATOR_ID
+    updater = Account(name="Updater", email="updater@example.com")
+    stored_features = json.dumps(
+        {
+            "opening_statement": "Workflow opening",
+            "file_upload": {"image": {"enabled": True, "number_limits": 3, "transfer_methods": ["local_file"]}},
+        }
+    )
+    with sqlite_session_factory.begin() as session:
+        session.add_all([author, updater])
+        session.flush()
+        workflow = Workflow(
+            tenant_id=preview_ref.tenant_id,
+            app_id=preview_ref.app_id,
+            type=WorkflowType.CHAT,
+            version="published",
+            graph='{"nodes":[],"edges":[]}',
+            features=stored_features,
+            created_by=author.id,
+            updated_by=updater.id,
+            marked_name="Release",
+            marked_comment="Preview snapshot",
+        )
+        workflow.created_at = _CREATED_AT
+        workflow.updated_at = _CREATED_AT
+        workflow._environment_variables = stored_environment
+        workflow.conversation_variables = [
+            StringVariable(id="topic", name="topic", value="sqlite", selector=["conversation", "topic"])
+        ]
+        session.add(workflow)
+        session.flush()
+        app = session.get(App, preview_ref.app_id)
+        assert app is not None
+        app.workflow_id = workflow.id
+        session.add(
+            WorkflowToolProvider(
+                name="preview-workflow",
+                label="Preview workflow",
+                icon="icon",
+                app_id=preview_ref.app_id,
+                version="1.0.0",
+                user_id=author.id,
+                tenant_id=preview_ref.tenant_id,
+                description="Workflow provider",
+                parameter_configuration="[]",
+            )
+        )
+
+    closed: list[Session] = []
+    mutated: list[object] = []
+
+    class TrackedSession(Session):
+        @override
+        def close(self) -> None:
+            mutated.extend(self.new)
+            mutated.extend(self.dirty)
+            mutated.extend(self.deleted)
+            super().close()
+            closed.append(self)
+
+    factory: sessionmaker[Session] = sessionmaker(bind=sqlite_engine, class_=TrackedSession, expire_on_commit=False)
+    record = AppPreviewQueryRepository(session_factory=factory).get_workflow(app=preview_ref)
+
+    assert isinstance(sqlite_engine.pool, QueuePool)
+    assert sqlite_engine.pool.checkedout() == 0
+    assert closed
+    assert all(not session.in_transaction() and not session.identity_map for session in closed)
+    assert mutated == []
+    assert record.tenant_id == preview_ref.tenant_id
+    assert record.environment_variables_json == stored_environment
+    result = record.workflow
+    assert result.environment_variables == ()
+    assert result.id == workflow.id
+    assert result.graph == {"nodes": [], "edges": []}
+    assert result.features == {
+        "opening_statement": "Workflow opening",
+        "file_upload": {
+            "enabled": True,
+            "number_limits": 3,
+            "allowed_file_upload_methods": ["local_file"],
+            "allowed_file_types": ["image"],
+            "allowed_file_extensions": [],
+        },
+    }
+    assert result.hash == workflow.unique_hash
+    assert result.version == "published"
+    assert result.marked_name == "Release"
+    assert result.marked_comment == "Preview snapshot"
+    assert result.created_by == AppPreviewAccount(id=author.id, name="Author", email="author@example.com")
+    assert result.updated_by == AppPreviewAccount(id=updater.id, name="Updater", email="updater@example.com")
+    assert result.created_at == _CREATED_AT
+    assert result.updated_at == _CREATED_AT
+    assert result.tool_published is True
+    assert len(result.conversation_variables) == 1
+    assert result.conversation_variables[0]["name"] == "topic"
+    assert result.conversation_variables[0]["value"] == "sqlite"
+    assert result.conversation_variables[0]["value_type"] == "string"
+    assert result.rag_pipeline_variables == ()
+
+    def serialize_date(value: object) -> str:
+        assert isinstance(value, datetime), f"Non-data value escaped repository: {type(value)}"
+        return value.isoformat()
+
+    json.dumps(asdict(record), default=serialize_date)
+    with sqlite_session_factory() as session:
+        assert session.scalar(select(Workflow._features).where(Workflow.id == workflow.id)) == stored_features
+        assert (
+            session.scalar(select(Workflow._environment_variables).where(Workflow.id == workflow.id))
+            == stored_environment
+        )

api/tests/unit_tests/services/test_app_preview_details_adapters.py

@@ -0,0 +1,649 @@
+"""Real preview detail reads preserve ownership, masking and transaction boundaries."""
+
+import json
+from collections.abc import Iterator, Sequence
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from typing import Literal, cast, override
+from uuid import uuid4
+
+import pytest
+from flask import Flask
+from pydantic import JsonValue
+from sqlalchemy import Connection, Engine, ExecutionContext, delete, event, select, text, update
+from sqlalchemy.orm import Session, SessionTransaction, UOWTransaction, sessionmaker
+from sqlalchemy.pool import QueuePool
+
+import core.agent.tool_configuration as tool_configuration_module
+import core.helper.tool_parameter_cache as cache_module
+from core.agent.entities import AgentToolEntity
+from core.helper import encrypter
+from core.plugin.plugin_service import PluginService
+from core.tools.__base.tool import Tool
+from core.tools.entities.tool_entities import ApiProviderSchemaType, ToolEntity, ToolParameter
+from core.tools.tool_manager import ToolManager
+from extensions.ext_database import db
+from graphon.variables import SecretVariable
+from models import Account, App
+from models.account import Tenant
+from models.enums import CustomizeTokenStrategy
+from models.model import AppMode, AppModelConfig, Site
+from models.provider_ids import GenericProviderID
+from models.tools import ApiToolProvider
+from models.workflow import Workflow, WorkflowType
+from repositories.app_preview_query_repository import AppPreviewQueryRepository
+from services.account_errors import AccountNotFoundError
+from services.app_definition_query_service import AppDefinitionUnavailableError
+from services.app_preview_details_adapters import AppPreviewDetailsRuntime
+from services.app_preview_details_service import AppPreviewDetail
+from services.app_preview_query_service import AppPreviewRef, AppPreviewSiteUnavailableError, AppPreviewUnavailableError
+
+
+@dataclass(frozen=True)
+class _Harness:
+    flask_app: Flask
+    target: App
+    decoy: App
+    account: Account
+    active_workspace: Tenant
+    config: AppModelConfig
+    decoy_config: AppModelConfig
+    workflow: Workflow
+    decoy_workflow: Workflow
+    factory: sessionmaker[Session]
+    engine: Engine
+    runtime: AppPreviewDetailsRuntime
+    closed: list[Session]
+    mutated: list[object]
+    commits: list[Session]
+    writes: list[str]
+
+    @property
+    def reference(self) -> AppPreviewRef:
+        return AppPreviewRef(app_id=self.target.id, tenant_id=self.target.tenant_id)
+
+    def detail(self, *, app: AppPreviewRef | None = None, account_id: str | None = None) -> AppPreviewDetail:
+        return self.runtime.get_detail(
+            app=app or self.reference,
+            account_id=account_id or self.account.id,
+            active_workspace_id=self.active_workspace.id,
+        )
+
+    def assert_closed(self) -> None:
+        assert isinstance(self.engine.pool, QueuePool)
+        assert self.engine.pool.checkedout() == 0
+        assert self.closed
+        assert all(not session.in_transaction() and not session.identity_map for session in self.closed)
+
+    def assert_read_only(self) -> None:
+        self.assert_closed()
+        assert self.mutated == []
+        assert self.commits == []
+        assert self.writes == []
+
+
+@pytest.fixture
+def harness(sqlite_engine: Engine, sqlite_session_factory: sessionmaker[Session]) -> Iterator[_Harness]:
+    target = App(
+        id=str(uuid4()), tenant_id=str(uuid4()), name="Preview", mode="chat", enable_site=True, enable_api=True
+    )
+    decoy = App(id=str(uuid4()), tenant_id=str(uuid4()), name="Decoy", mode="chat", enable_site=True, enable_api=True)
+    account = Account(name="Viewer", email="preview@example.com")
+    active_workspace = Tenant(name="Active workspace")
+    config = AppModelConfig(app_id=target.id, opening_statement="Preview opening", pre_prompt="Visible prompt")
+    decoy_config = AppModelConfig(app_id=decoy.id, opening_statement="Other tenant prompt")
+    with sqlite_session_factory.begin() as session:
+        session.add_all([target, decoy, account, active_workspace, config, decoy_config])
+        session.flush()
+        target.app_model_config_id = config.id
+        decoy.app_model_config_id = decoy_config.id
+        workflow = Workflow(
+            tenant_id=target.tenant_id,
+            app_id=target.id,
+            type=WorkflowType.CHAT,
+            version="published",
+            graph='{"nodes":[],"edges":[]}',
+            features='{"opening_statement":"Workflow opening"}',
+            created_by=account.id,
+            updated_by=account.id,
+        )
+        decoy_workflow = Workflow(
+            tenant_id=decoy.tenant_id,
+            app_id=decoy.id,
+            type=WorkflowType.CHAT,
+            version="published",
+            graph='{"nodes":[],"edges":[]}',
+            features='{"opening_statement":"Other tenant workflow"}',
+            created_by=account.id,
+        )
+        session.add_all([workflow, decoy_workflow])
+        session.flush()
+        target.workflow_id = workflow.id
+        decoy.workflow_id = decoy_workflow.id
+        session.add_all(
+            [
+                Site(
+                    app_id=target.id,
+                    title="Preview site",
+                    default_language="en-US",
+                    customize_token_strategy=CustomizeTokenStrategy.UUID,
+                ),
+                Site(
+                    app_id=decoy.id,
+                    title="Decoy site",
+                    default_language="en-US",
+                    customize_token_strategy=CustomizeTokenStrategy.UUID,
+                ),
+            ]
+        )
+    closed: list[Session] = []
+    mutated: list[object] = []
+    commits: list[Session] = []
+    writes: list[str] = []
+
+    def record_changes(session: Session) -> None:
+        mutated.extend(session.new)
+        mutated.extend(session.dirty)
+        mutated.extend(session.deleted)
+
+    class TrackedSession(Session):
+        @override
+        def close(self) -> None:
+            record_changes(self)
+            super().close()
+            closed.append(self)
+
+    factory: sessionmaker[Session] = sessionmaker(bind=sqlite_engine, class_=TrackedSession, expire_on_commit=False)
+
+    @event.listens_for(factory, "before_flush")
+    def before_flush(session: Session, _context: UOWTransaction, _instances: Sequence[object] | None) -> None:
+        record_changes(session)
+
+    @event.listens_for(factory, "before_commit")
+    def before_commit(session: Session) -> None:
+        commits.append(session)
+
+    def before_cursor_execute(
+        _connection: Connection,
+        _cursor: object,
+        statement: str,
+        _parameters: object,
+        _context: ExecutionContext,
+        _executemany: bool,
+    ) -> None:
+        if statement.lstrip().split(maxsplit=1)[0].upper() in {"INSERT", "UPDATE", "DELETE"}:
+            writes.append(statement)
+
+    @event.listens_for(factory, "after_begin")
+    def after_begin(_session: Session, _transaction: SessionTransaction, connection: Connection) -> None:
+        event.listen(connection, "before_cursor_execute", before_cursor_execute)
+
+    flask_app = Flask(__name__)
+    flask_app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI=str(sqlite_engine.url))
+    db.init_app(flask_app)
+    yield _Harness(
+        flask_app,
+        target,
+        decoy,
+        account,
+        active_workspace,
+        config,
+        decoy_config,
+        workflow,
+        decoy_workflow,
+        sqlite_session_factory,
+        sqlite_engine,
+        AppPreviewDetailsRuntime(details=AppPreviewQueryRepository(session_factory=factory)),
+        closed,
+        mutated,
+        commits,
+        writes,
+    )
+    with flask_app.app_context():
+        db.session.remove()
+        db.engine.dispose()
+
+
+def _serialize_date(value: object) -> str:
+    assert isinstance(value, datetime), f"Non-data value escaped runtime: {type(value)}"
+    return value.isoformat()
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflow"])
+def test_preview_result_contains_only_detached_data(harness: _Harness, endpoint: Literal["detail", "workflow"]) -> None:
+    with harness.flask_app.app_context():
+        result = harness.detail() if endpoint == "detail" else harness.runtime.get_workflow(app=harness.reference)
+    assert result.id == (harness.target.id if endpoint == "detail" else harness.workflow.id)
+    json.dumps(asdict(result), default=_serialize_date)
+    if isinstance(result, AppPreviewDetail):
+        assert result.site.title == "Preview site"
+        assert result.model_config is not None
+        assert result.model_config["opening_statement"] == "Preview opening"
+        assert result.workflow is not None
+        assert result.workflow.id == harness.workflow.id
+    else:
+        assert result.features["opening_statement"] == "Workflow opening"
+        assert result.created_by is not None
+        assert result.created_by.id == harness.account.id
+        assert result.updated_by == result.created_by
+    harness.assert_read_only()
+
+
+@pytest.mark.parametrize("plugin_failure", [False, True])
+def test_deleted_tools_check_owner_scoped_provider_snapshot_after_session_closes(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, plugin_failure: bool
+) -> None:
+    own_provider = ApiToolProvider(
+        name="Owner API",
+        icon="icon",
+        schema="{}",
+        schema_type_str=ApiProviderSchemaType.OPENAPI,
+        user_id=harness.account.id,
+        tenant_id=harness.target.tenant_id,
+        description="Owner provider",
+        tools_str="[]",
+        credentials_str="{}",
+    )
+    other_provider = ApiToolProvider(
+        name="Viewer API",
+        icon="icon",
+        schema="{}",
+        schema_type_str=ApiProviderSchemaType.OPENAPI,
+        user_id=harness.account.id,
+        tenant_id=harness.active_workspace.id,
+        description="Provider from another tenant",
+        tools_str="[]",
+        credentials_str="{}",
+    )
+    missing_api_id = str(uuid4())
+    configured_tools = [
+        ("builtin", "acme/missing/missing", "missing-plugin"),
+        ("builtin", "local", "hardcoded"),
+        ("api", other_provider.id, "other-tenant-api"),
+        ("builtin", "acme/present/present", "present-plugin"),
+        ("api", own_provider.id, "owner-api"),
+        ("api", missing_api_id, "missing-api"),
+    ]
+    with harness.factory.begin() as session:
+        session.add_all([own_provider, other_provider])
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(
+                agent_mode=json.dumps(
+                    {
+                        "enabled": False,
+                        "tools": [
+                            {
+                                "provider_type": kind,
+                                "provider_id": provider_id,
+                                "tool_name": name,
+                                "tool_parameters": {},
+                            }
+                            for kind, provider_id, name in configured_tools
+                        ],
+                    }
+                )
+            )
+        )
+    calls: list[tuple[str, tuple[str, ...]]] = []
+
+    def hardcoded_provider(provider_id: str) -> object:
+        harness.assert_closed()
+        if provider_id == "local":
+            return object()
+        raise KeyError(provider_id)
+
+    def check_plugins(tenant_id: str, provider_ids: Sequence[GenericProviderID]) -> Sequence[bool]:
+        harness.assert_read_only()
+        calls.append((tenant_id, tuple(str(provider_id) for provider_id in provider_ids)))
+        if plugin_failure:
+            raise RuntimeError("Plugin daemon unavailable")
+        return [False, True]
+
+    monkeypatch.setattr(ToolManager, "get_hardcoded_provider", hardcoded_provider)
+    monkeypatch.setattr(PluginService, "check_tools_existence", check_plugins)
+    with harness.flask_app.app_context():
+        if plugin_failure:
+            with pytest.raises(RuntimeError, match="Plugin daemon unavailable"):
+                harness.detail()
+        else:
+            detail = harness.detail()
+            assert [asdict(tool) for tool in detail.deleted_tools] == [
+                {"type": "builtin", "tool_name": "missing-plugin", "provider_id": "acme/missing/missing"},
+                {"type": "api", "tool_name": "other-tenant-api", "provider_id": other_provider.id},
+                {"type": "api", "tool_name": "missing-api", "provider_id": missing_api_id},
+            ]
+    assert calls == [(harness.target.tenant_id, ("acme/missing/missing", "acme/present/present"))]
+    harness.assert_read_only()
+
+
+@pytest.mark.parametrize(
+    "malformed_fields",
+    [
+        {},
+        {"provider_id": None},
+        {"provider_id": 1},
+        {"provider_id": {}},
+        {"provider_id": ""},
+        {"provider_id": "invalid/provider-id"},
+    ],
+    ids=["missing", "null", "number", "object", "empty", "invalid-format"],
+)
+def test_deleted_tools_ignore_invalid_builtin_config_and_keep_valid_missing_provider(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, malformed_fields: dict[str, JsonValue]
+) -> None:
+    configured_tools: list[dict[str, JsonValue]] = [
+        {
+            "provider_type": "builtin",
+            "tool_name": "invalid-plugin",
+            "tool_parameters": {},
+            "enabled": True,
+            **malformed_fields,
+        },
+        {
+            "provider_type": "builtin",
+            "provider_id": "acme/missing/missing",
+            "tool_name": "missing-plugin",
+            "tool_parameters": {},
+        },
+    ]
+    with harness.factory.begin() as session:
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(agent_mode=json.dumps({"enabled": False, "tools": configured_tools}))
+        )
+    calls: list[tuple[str, tuple[str, ...]]] = []
+
+    def hardcoded_provider(provider_id: str) -> object:
+        harness.assert_closed()
+        raise KeyError(provider_id)
+
+    def check_plugins(tenant_id: str, provider_ids: Sequence[GenericProviderID]) -> Sequence[bool]:
+        harness.assert_read_only()
+        calls.append((tenant_id, tuple(str(provider_id) for provider_id in provider_ids)))
+        return [False]
+
+    monkeypatch.setattr(ToolManager, "get_hardcoded_provider", hardcoded_provider)
+    monkeypatch.setattr(PluginService, "check_tools_existence", check_plugins)
+    with harness.flask_app.app_context():
+        detail = harness.detail()
+
+    assert calls == [(harness.target.tenant_id, ("acme/missing/missing",))]
+    assert [asdict(tool) for tool in detail.deleted_tools] == [
+        {"type": "builtin", "tool_name": "missing-plugin", "provider_id": "acme/missing/missing"}
+    ]
+    assert detail.model_config is not None
+    agent_mode = detail.model_config["agent_mode"]
+    assert isinstance(agent_mode, dict)
+    assert agent_mode["tools"] == configured_tools
+    harness.assert_read_only()
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflow"])
+@pytest.mark.parametrize("change", ["missing", "tenant", "status"])
+def test_app_scope_never_falls_back_to_another_valid_app(
+    harness: _Harness, endpoint: Literal["detail", "workflow"], change: Literal["missing", "tenant", "status"]
+) -> None:
+    reference = harness.reference
+    if change == "missing":
+        reference = AppPreviewRef(app_id=str(uuid4()), tenant_id=reference.tenant_id)
+    elif change == "tenant":
+        reference = AppPreviewRef(app_id=reference.app_id, tenant_id=harness.decoy.tenant_id)
+    else:
+        with harness.factory.begin() as session:
+            session.execute(
+                text("UPDATE apps SET status = 'disabled' WHERE id = :app_id"), {"app_id": reference.app_id}
+            )
+    get = harness.detail if endpoint == "detail" else harness.runtime.get_workflow
+    with harness.flask_app.app_context(), pytest.raises(AppPreviewUnavailableError):
+        get(app=reference)
+    harness.assert_closed()
+
+
+def test_detail_rejects_missing_account(harness: _Harness) -> None:
+    account_id = str(uuid4())
+    with harness.flask_app.app_context(), pytest.raises(AccountNotFoundError, match=account_id):
+        harness.detail(account_id=account_id)
+    harness.assert_closed()
+
+
+def test_detail_rejects_missing_site(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(Site).where(Site.app_id == harness.target.id))
+    with harness.flask_app.app_context(), pytest.raises(AppPreviewSiteUnavailableError):
+        harness.detail()
+    harness.assert_closed()
+
+
+def test_detail_rejects_config_link_to_another_app(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(
+            update(App).where(App.id == harness.target.id).values({App.app_model_config_id: harness.decoy_config.id})
+        )
+    with harness.flask_app.app_context(), pytest.raises(AppDefinitionUnavailableError):
+        harness.detail()
+    harness.assert_closed()
+
+
+def test_detail_missing_model_config_is_optional(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(AppModelConfig).where(AppModelConfig.id == harness.config.id))
+    with harness.flask_app.app_context():
+        result = harness.detail()
+    assert result.model_config is None
+    harness.assert_closed()
+
+
+def test_detail_closes_session_if_model_config_cannot_be_decoded(harness: _Harness) -> None:
+    with harness.factory.begin() as session:
+        session.execute(update(AppModelConfig).where(AppModelConfig.id == harness.config.id).values(model="{"))
+    with harness.flask_app.app_context(), pytest.raises(json.JSONDecodeError):
+        harness.detail()
+    harness.assert_closed()
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflow"])
+@pytest.mark.parametrize("wrong_owner", ["app", "tenant"])
+def test_workflow_link_checks_both_app_and_tenant(
+    harness: _Harness, endpoint: Literal["detail", "workflow"], wrong_owner: Literal["app", "tenant"]
+) -> None:
+    values = {"app_id": harness.decoy.id} if wrong_owner == "app" else {"tenant_id": harness.decoy.tenant_id}
+    with harness.factory.begin() as session:
+        session.execute(update(Workflow).where(Workflow.id == harness.workflow.id).values(values))
+    get = harness.detail if endpoint == "detail" else harness.runtime.get_workflow
+    with harness.flask_app.app_context(), pytest.raises(AppDefinitionUnavailableError):
+        get(app=harness.reference)
+    harness.assert_closed()
+
+
+@pytest.mark.parametrize("endpoint", ["detail", "workflow"])
+def test_missing_workflow_is_optional_only_in_app_detail(
+    harness: _Harness, endpoint: Literal["detail", "workflow"]
+) -> None:
+    with harness.factory.begin() as session:
+        session.execute(delete(Workflow).where(Workflow.id == harness.workflow.id))
+    with harness.flask_app.app_context():
+        if endpoint == "detail":
+            assert harness.detail().workflow is None
+        else:
+            with pytest.raises(AppDefinitionUnavailableError):
+                harness.runtime.get_workflow(app=harness.reference)
+    harness.assert_closed()
+
+
+@pytest.mark.parametrize("legacy_agent", [False, True])
+def test_detail_computes_legacy_mode_without_writes_or_committing_callers_session(
+    harness: _Harness, legacy_agent: bool
+) -> None:
+    if legacy_agent:
+        with harness.factory.begin() as session:
+            session.execute(
+                update(AppModelConfig)
+                .where(AppModelConfig.id == harness.config.id)
+                .values(agent_mode=json.dumps({"enabled": True, "strategy": "react", "tools": []}))
+            )
+    with harness.flask_app.app_context():
+        caller = db.session()
+        account = caller.get(Account, harness.account.id)
+        assert account is not None
+        account.name = "Uncommitted caller edit"
+        result = harness.detail()
+        assert db.session() is caller
+        assert account in caller.dirty
+        assert result.mode == (AppMode.AGENT_CHAT if legacy_agent else AppMode.CHAT)
+        harness.assert_read_only()
+        with harness.factory() as session:
+            assert session.scalar(select(Account.name).where(Account.id == account.id)) == "Viewer"
+            assert session.scalar(select(App.mode).where(App.id == harness.target.id)) == AppMode.CHAT
+
+
+@pytest.mark.parametrize("failure", [False, True])
+def test_workflow_secret_decryption_happens_after_query_session_closes(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, failure: bool
+) -> None:
+    secret = SecretVariable(id=str(uuid4()), name="api_key", value="encrypted", selector=["env", "api_key"])
+    with harness.factory.begin() as session:
+        session.execute(
+            update(Workflow)
+            .where(Workflow.id == harness.workflow.id)
+            .values({Workflow._environment_variables: json.dumps({secret.name: secret.model_dump(mode="json")})})
+        )
+    calls: list[str] = []
+
+    def decrypt(tenant_id: str, token: str) -> str:
+        harness.assert_closed()
+        assert token == "encrypted"
+        calls.append(tenant_id)
+        if failure:
+            raise RuntimeError("Key service unavailable")
+        return "decrypted-secret"
+
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+    with harness.flask_app.app_context():
+        if failure:
+            with pytest.raises(RuntimeError, match="Key service unavailable"):
+                harness.runtime.get_workflow(app=harness.reference)
+        else:
+            result = harness.runtime.get_workflow(app=harness.reference)
+            assert result.environment_variables[0]["value"] == "decrypted-secret"
+            assert result.environment_variables[0]["value_type"] == "secret"
+    assert calls == [harness.target.tenant_id]
+    harness.assert_closed()
+
+
+@dataclass
+class _ExternalTool:
+    entity: ToolEntity
+
+    def get_runtime_parameters(self) -> list[ToolParameter]:
+        return []
+
+
+class _EmptyCache:
+    def get(self, key: str) -> None:
+        assert key
+
+    def setex(self, key: str, expiry: int, value: str) -> None:
+        assert key
+        assert expiry > 0
+        assert value
+
+
+@pytest.mark.parametrize("failure", ["none", "runtime", "decrypt"])
+def test_agent_tool_masking_returns_a_copy_without_dirtying_or_flushing_models(
+    harness: _Harness, monkeypatch: pytest.MonkeyPatch, failure: Literal["none", "runtime", "decrypt"]
+) -> None:
+    provider_id = str(uuid4())
+    agent_mode = {
+        "enabled": True,
+        "strategy": "react",
+        "tools": [
+            {
+                "provider_type": "workflow",
+                "provider_id": provider_id,
+                "tool_name": "search",
+                "tool_parameters": {"api_key": "encrypted"},
+            }
+        ],
+    }
+    with harness.factory.begin() as session:
+        session.execute(update(App).where(App.id == harness.target.id).values(mode=AppMode.AGENT_CHAT))
+        session.execute(
+            update(AppModelConfig)
+            .where(AppModelConfig.id == harness.config.id)
+            .values(agent_mode=json.dumps(agent_mode))
+        )
+    tool = _ExternalTool(
+        ToolEntity.model_validate(
+            {
+                "identity": {"author": "test", "name": "search", "label": {"en_US": "Search"}, "provider": provider_id},
+                "parameters": [{"name": "api_key", "label": {"en_US": "Key"}, "type": "secret-input", "form": "form"}],
+            }
+        )
+    )
+    calls: list[str] = []
+    checkedout_during_io: list[int] = []
+
+    def runtime(*, tenant_id: str, app_id: str, agent_tool: AgentToolEntity, user_id: str) -> Tool:
+        assert isinstance(harness.engine.pool, QueuePool)
+        checkedout_during_io.append(harness.engine.pool.checkedout())
+        harness.assert_closed()
+        assert tenant_id == harness.active_workspace.id
+        assert tenant_id != harness.target.tenant_id
+        assert app_id == harness.target.id
+        assert user_id == harness.account.id
+        assert agent_tool.tool_name == "search"
+        calls.append(tenant_id)
+        if failure == "runtime":
+            raise RuntimeError("Tool provider unavailable")
+        return cast(Tool, tool)
+
+    def decrypt(tenant_id: str, token: str) -> str:
+        assert isinstance(harness.engine.pool, QueuePool)
+        checkedout_during_io.append(harness.engine.pool.checkedout())
+        harness.assert_closed()
+        assert tenant_id == harness.active_workspace.id
+        assert token == "encrypted"
+        if failure == "decrypt":
+            raise RuntimeError("Credential provider unavailable")
+        return "secret-value"
+
+    monkeypatch.setattr(tool_configuration_module.ToolManager, "get_agent_tool_runtime", runtime)
+    monkeypatch.setattr(cache_module, "redis_client", _EmptyCache())
+    monkeypatch.setattr(encrypter, "decrypt_token", decrypt)
+    original_agent_mode = json.dumps(agent_mode)
+    with harness.flask_app.app_context():
+        result = harness.detail()
+        masked_copy = tool_configuration_module.mask_agent_tool_parameters(
+            agent_mode=cast(dict[str, JsonValue], agent_mode),
+            app_id=harness.target.id,
+            tenant_id=harness.active_workspace.id,
+            user_id=harness.account.id,
+        )
+    assert result.model_config is not None
+    masked_agent_mode = result.model_config["agent_mode"]
+    assert isinstance(masked_agent_mode, dict)
+    tools = masked_agent_mode["tools"]
+    assert isinstance(tools, list)
+    masked_tool = tools[0]
+    assert isinstance(masked_tool, dict)
+    # The credential helper can fall back to ciphertext, which is still masked.
+    # Without a tool runtime, the response must omit unverified parameter values.
+    expected_parameters: dict[str, dict[str, str]] = {
+        "none": {"api_key": "se********ue"},
+        "runtime": {},
+        "decrypt": {"api_key": "en*****ed"},
+    }
+    assert masked_tool["tool_parameters"] == expected_parameters[failure]
+    assert masked_copy == masked_agent_mode
+    assert masked_copy is not agent_mode
+    assert json.dumps(agent_mode) == original_agent_mode
+    assert calls == [harness.active_workspace.id, harness.active_workspace.id]
+    assert checkedout_during_io == ([0, 0] if failure == "runtime" else [0, 0, 0, 0])
+    with harness.factory() as session:
+        persisted = session.get(AppModelConfig, harness.config.id)
+        assert persisted is not None
+        assert persisted.agent_mode_dict == agent_mode
+    harness.assert_read_only()

packages/contracts/generated/api/console/trial-apps/orpc.gen.ts

@@ -310,7 +310,7 @@ export const tasks = {
 }
 
 /**
- * Get workflow detail
+ * Get a detached workflow definition after catalog preview admission
  */
 export const get5 = oc
   .route({
@@ -318,7 +318,7 @@ export const get5 = oc
     method: 'GET',
     operationId: 'getTrialAppsByAppIdWorkflows',
     path: '/trial-apps/{app_id}/workflows',
-    summary: 'Get workflow detail',
+    summary: 'Get a detached workflow definition after catalog preview admission',
     tags: ['console'],
   })
   .input(z.object({ params: zGetTrialAppsByAppIdWorkflowsPath }))
@@ -331,7 +331,7 @@ export const workflows = {
 }
 
 /**
- * Get app detail
+ * Get app detail using the viewer's workspace for tool configuration
  */
 export const get6 = oc
   .route({
@@ -339,7 +339,7 @@ export const get6 = oc
     method: 'GET',
     operationId: 'getTrialAppsByAppId',
     path: '/trial-apps/{app_id}',
-    summary: 'Get app detail',
+    summary: "Get app detail using the viewer's workspace for tool configuration",
     tags: ['console'],
   })
   .input(z.object({ params: zGetTrialAppsByAppIdPath }))