refactor(web): remove annotation control React.FC

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

api/.importlinter

@@ -95,3 +95,33 @@ forbidden_modules =
     services.workspace_member_role_resolver
     sqlalchemy
     werkzeug
+
+[importlinter:contract:setup-service-boundary]
+name = Setup application service is framework and persistence neutral
+type = forbidden
+source_modules =
+    services.setup_service
+forbidden_modules =
+    configs
+    controllers
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    werkzeug
+
+[importlinter:contract:schema-definition-service-boundary]
+name = Schema definition application service is framework and persistence neutral
+type = forbidden
+source_modules =
+    services.schema_definition_service
+forbidden_modules =
+    configs
+    controllers
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    werkzeug

api/controllers/console/app/agent_config_inspector.py

@@ -30,6 +30,7 @@ from controllers.console.wraps import (
     RBACResourceScope,
     account_initialization_required,
     edit_permission_required,
+    model_validate,
     rbac_permission_required,
     setup_required,
     with_current_tenant_id,
@@ -810,14 +811,21 @@ class AgentConfigFilesByAgentApi(Resource):
     @with_current_user
     @with_current_tenant_id
     @with_session
-    def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
-        payload = AgentConfigFileUploadPayload.model_validate(console_ns.payload or {})
+    @model_validate(AgentConfigFileUploadPayload)
+    def post(
+        self,
+        req_data: AgentConfigFileUploadPayload,
+        session: Session,
+        tenant_id: str,
+        current_user: Account,
+        agent_id: UUID,
+    ):
         return _with_agent_route_target(
             session=session,
             tenant_id=tenant_id,
             agent_id=agent_id,
             current_user=current_user,
-            action=lambda target: _file_upload_response(target, payload),
+            action=lambda target: _file_upload_response(target, req_data),
         )
 
 
@@ -849,13 +857,13 @@ class AgentConfigFilesApi(Resource):
     @with_current_user
     @with_session
     @get_app_model(mode=_WORKFLOW_APP_MODES)
-    def post(self, session: Session, current_user: Account, app_model: App):
-        payload = AgentConfigFileUploadPayload.model_validate(console_ns.payload or {})
+    @model_validate(AgentConfigFileUploadPayload)
+    def post(self, req_data: AgentConfigFileUploadPayload, session: Session, current_user: Account, app_model: App):
         return _with_app_route_target(
             session=session,
             app_model=app_model,
             current_user=current_user,
-            action=lambda target: _file_upload_response(target, payload),
+            action=lambda target: _file_upload_response(target, req_data),
         )
 
 

api/controllers/console/app/annotation.py

@@ -20,6 +20,7 @@ from controllers.console.wraps import (
     annotation_import_rate_limit,
     cloud_edition_billing_resource_check,
     edit_permission_required,
+    model_validate,
     rbac_permission_required,
     setup_required,
 )
@@ -180,14 +181,14 @@ class AnnotationReplyActionApi(Resource):
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
-    def post(self, app_id: UUID, action: Literal["enable", "disable"]):
-        args = AnnotationReplyPayload.model_validate(console_ns.payload)
+    @model_validate(AnnotationReplyPayload)
+    def post(self, req_data: AnnotationReplyPayload, app_id: UUID, action: Literal["enable", "disable"]):
         match action:
             case "enable":
                 enable_args: EnableAnnotationArgs = {
-                    "score_threshold": args.score_threshold,
-                    "embedding_provider_name": args.embedding_provider_name,
-                    "embedding_model_name": args.embedding_model_name,
+                    "score_threshold": req_data.score_threshold,
+                    "embedding_provider_name": req_data.embedding_provider_name,
+                    "embedding_model_name": req_data.embedding_model_name,
                 }
                 result = AppAnnotationService.enable_app_annotation(enable_args, str(app_id))
             case "disable":
@@ -231,12 +232,17 @@ class AppAnnotationSettingUpdateApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_session
-    def post(self, session: Session, app_id: UUID, annotation_setting_id: UUID):
+    @model_validate(AnnotationSettingUpdatePayload)
+    def post(
+        self,
+        req_data: AnnotationSettingUpdatePayload,
+        session: Session,
+        app_id: UUID,
+        annotation_setting_id: UUID,
+    ):
         annotation_setting_id_str = str(annotation_setting_id)
 
-        args = AnnotationSettingUpdatePayload.model_validate(console_ns.payload)
-
-        setting_args: UpdateAnnotationSettingArgs = {"score_threshold": args.score_threshold}
+        setting_args: UpdateAnnotationSettingArgs = {"score_threshold": req_data.score_threshold}
         result = AppAnnotationService.update_app_annotation_setting(
             str(app_id), annotation_setting_id_str, setting_args, session
         )
@@ -290,11 +296,11 @@ class AnnotationApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     @with_session(write=False)
-    def get(self, session: Session, app_id: UUID):
-        args = AnnotationListQuery.model_validate(request.args.to_dict(flat=True))
-        page = args.page
-        limit = args.limit
-        keyword = args.keyword
+    @model_validate(AnnotationListQuery)
+    def get(self, req_data: AnnotationListQuery, session: Session, app_id: UUID):
+        page = req_data.page
+        limit = req_data.limit
+        keyword = req_data.keyword
 
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
             str(app_id), page, limit, keyword, session
@@ -317,17 +323,17 @@ class AnnotationApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_session
-    def post(self, session: Session, app_id: UUID):
-        args = CreateAnnotationPayload.model_validate(console_ns.payload)
+    @model_validate(CreateAnnotationPayload)
+    def post(self, req_data: CreateAnnotationPayload, session: Session, app_id: UUID):
         upsert_args: UpsertAnnotationArgs = {}
-        if args.answer is not None:
-            upsert_args["answer"] = args.answer
-        if args.content is not None:
-            upsert_args["content"] = args.content
-        if args.message_id is not None:
-            upsert_args["message_id"] = args.message_id
-        if args.question is not None:
-            upsert_args["question"] = args.question
+        if req_data.answer is not None:
+            upsert_args["answer"] = req_data.answer
+        if req_data.content is not None:
+            upsert_args["content"] = req_data.content
+        if req_data.message_id is not None:
+            upsert_args["message_id"] = req_data.message_id
+        if req_data.question is not None:
+            upsert_args["question"] = req_data.question
         annotation = AppAnnotationService.up_insert_app_annotation_from_message(upsert_args, str(app_id), session)
         return dump_response(Annotation, annotation), 201
 
@@ -407,13 +413,13 @@ class AnnotationUpdateDeleteApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_session
-    def post(self, session: Session, app_id: UUID, annotation_id: UUID):
-        args = UpdateAnnotationPayload.model_validate(console_ns.payload)
+    @model_validate(UpdateAnnotationPayload)
+    def post(self, req_data: UpdateAnnotationPayload, session: Session, app_id: UUID, annotation_id: UUID):
         update_args: UpdateAnnotationArgs = {}
-        if args.answer is not None:
-            update_args["answer"] = args.answer
-        if args.question is not None:
-            update_args["question"] = args.question
+        if req_data.answer is not None:
+            update_args["answer"] = req_data.answer
+        if req_data.question is not None:
+            update_args["question"] = req_data.question
         app_ref = _get_app_ref(session, str(app_id))
         annotation_ref = AppRefService.create_annotation_ref(app_ref, str(annotation_id))
         annotation = AppAnnotationService.update_app_annotation_directly(update_args, annotation_ref, session)

api/controllers/console/app/ops_trace.py

@@ -1,6 +1,5 @@
 from typing import Any
 
-from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import BadRequest
@@ -14,6 +13,7 @@ from controllers.console.wraps import (
     RBACResourceScope,
     account_initialization_required,
     edit_permission_required,
+    model_validate,
     rbac_permission_required,
     setup_required,
 )
@@ -74,12 +74,11 @@ class TraceAppConfigApi(Resource):
     @account_initialization_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
-    def get(self, app_model: App):
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-
+    @model_validate(TraceProviderQuery)
+    def get(self, req_data: TraceProviderQuery, app_model: App):
         try:
             trace_config = OpsService.get_tracing_app_config(
-                app_id=app_model.id, tracing_provider=args.tracing_provider, session=db.session()
+                app_id=app_model.id, tracing_provider=req_data.tracing_provider, session=db.session()
             )
             if not trace_config:
                 return {"has_not_configured": True}
@@ -104,15 +103,14 @@ class TraceAppConfigApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
-    def post(self, app_model: App):
+    @model_validate(TraceConfigPayload)
+    def post(self, req_data: TraceConfigPayload, app_model: App):
         """Create a new trace app configuration"""
-        args = TraceConfigPayload.model_validate(console_ns.payload)
-
         try:
             result = OpsService.create_tracing_app_config(
                 app_id=app_model.id,
-                tracing_provider=args.tracing_provider,
-                tracing_config=args.tracing_config,
+                tracing_provider=req_data.tracing_provider,
+                tracing_config=req_data.tracing_config,
                 session=db.session(),
             )
             if not result:
@@ -140,15 +138,14 @@ class TraceAppConfigApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
-    def patch(self, app_model: App):
+    @model_validate(TraceConfigPayload)
+    def patch(self, req_data: TraceConfigPayload, app_model: App):
         """Update an existing trace app configuration"""
-        args = TraceConfigPayload.model_validate(console_ns.payload)
-
         try:
             result = OpsService.update_tracing_app_config(
                 app_id=app_model.id,
-                tracing_provider=args.tracing_provider,
-                tracing_config=args.tracing_config,
+                tracing_provider=req_data.tracing_provider,
+                tracing_config=req_data.tracing_config,
                 session=db.session(),
             )
             if not result:
@@ -170,13 +167,12 @@ class TraceAppConfigApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
-    def delete(self, app_model: App):
+    @model_validate(TraceProviderQuery)
+    def delete(self, req_data: TraceProviderQuery, app_model: App):
         """Delete an existing trace app configuration"""
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
-
         try:
             result = OpsService.delete_tracing_app_config(
-                app_id=app_model.id, tracing_provider=args.tracing_provider, session=db.session()
+                app_id=app_model.id, tracing_provider=req_data.tracing_provider, session=db.session()
             )
             if not result:
                 raise TracingConfigNotExist()

api/controllers/console/explore/audio.py

@@ -20,6 +20,7 @@ from controllers.console.app.error import (
     UnsupportedAudioTypeError,
 )
 from controllers.console.explore.wraps import InstalledAppResource
+from controllers.console.wraps import model_validate
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
@@ -100,16 +101,15 @@ class ChatAudioApi(InstalledAppResource):
 class ChatTextApi(InstalledAppResource):
     @console_ns.expect(console_ns.models[TextToAudioPayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[AudioBinaryResponse.__name__])
-    def post(self, installed_app: InstalledApp):
+    @model_validate(TextToAudioPayload)
+    def post(self, req_data: TextToAudioPayload, installed_app: InstalledApp):
         app_model = installed_app.app_with_session(session=db.session())
         if app_model is None:
             raise AppUnavailableError()
         try:
-            payload = TextToAudioPayload.model_validate(console_ns.payload or {})
-
-            message_id = payload.message_id
-            text = payload.text
-            voice = payload.voice
+            message_id = req_data.message_id
+            text = req_data.text
+            voice = req_data.voice
             message_ref = None
             if message_id:
                 current_user, _ = current_account_with_tenant()

api/controllers/console/explore/completion.py

@@ -20,7 +20,7 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import with_session
 from controllers.console.explore.error import NotChatAppError, NotCompletionAppError
 from controllers.console.explore.wraps import InstalledAppResource
-from controllers.console.wraps import with_current_user, with_current_user_id
+from controllers.console.wraps import model_validate, with_current_user, with_current_user_id
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import (
@@ -89,17 +89,23 @@ class CompletionApi(InstalledAppResource):
     @console_ns.response(200, "Success")
     @with_current_user
     @with_session
-    def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
+    @model_validate(CompletionMessageExplorePayload)
+    def post(
+        self,
+        req_data: CompletionMessageExplorePayload,
+        session: Session,
+        current_user: Account,
+        installed_app: InstalledApp,
+    ):
         app_model = installed_app.app_with_session(session=session)
         if app_model is None:
             raise AppUnavailableError()
         if app_model.mode != AppMode.COMPLETION:
             raise NotCompletionAppError()
 
-        payload = CompletionMessageExplorePayload.model_validate(console_ns.payload or {})
-        args = payload.model_dump(exclude_none=True)
+        args = req_data.model_dump(exclude_none=True)
 
-        streaming = payload.response_mode == "streaming"
+        streaming = req_data.response_mode == "streaming"
         args["auto_generate_name"] = False
 
         installed_app.last_used_at = naive_utc_now()
@@ -173,7 +179,8 @@ class ChatApi(InstalledAppResource):
     @console_ns.response(200, "Success")
     @with_current_user
     @with_session
-    def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
+    @model_validate(ChatMessagePayload)
+    def post(self, req_data: ChatMessagePayload, session: Session, current_user: Account, installed_app: InstalledApp):
         app_model = installed_app.app_with_session(session=session)
         if app_model is None:
             raise AppUnavailableError()
@@ -181,8 +188,7 @@ class ChatApi(InstalledAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        payload = ChatMessagePayload.model_validate(console_ns.payload or {})
-        args = payload.model_dump(exclude_none=True)
+        args = req_data.model_dump(exclude_none=True)
 
         args["auto_generate_name"] = False
 
@@ -191,10 +197,10 @@ class ChatApi(InstalledAppResource):
 
         try:
             # Eagerly validate conversation to avoid hanging on invalid conversation_id
-            if payload.conversation_id:
+            if req_data.conversation_id:
                 ConversationService.get_conversation(
                     app_model=app_model,
-                    conversation_id=payload.conversation_id,
+                    conversation_id=req_data.conversation_id,
                     user=current_user,
                     session=session,
                 )

api/controllers/console/explore/conversation.py

@@ -11,7 +11,7 @@ from controllers.common.schema import query_params_from_model, register_response
 from controllers.console.app.error import AppUnavailableError
 from controllers.console.explore.error import NotChatAppError
 from controllers.console.explore.wraps import InstalledAppResource
-from controllers.console.wraps import with_current_user
+from controllers.console.wraps import model_validate, with_current_user
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from fields.conversation_fields import (
@@ -133,7 +133,8 @@ class ConversationRenameApi(InstalledAppResource):
     @console_ns.expect(console_ns.models[ConversationRenamePayload.__name__])
     @console_ns.response(200, "Conversation renamed successfully", console_ns.models[SimpleConversation.__name__])
     @with_current_user
-    def post(self, current_user: Account, installed_app: InstalledApp, c_id: UUID):
+    @model_validate(ConversationRenamePayload)
+    def post(self, req_data: ConversationRenamePayload, current_user: Account, installed_app: InstalledApp, c_id: UUID):
         app_model = installed_app.app_with_session(session=db.session())
         if app_model is None:
             raise AppUnavailableError()
@@ -143,12 +144,10 @@ class ConversationRenameApi(InstalledAppResource):
 
         conversation_id = str(c_id)
 
-        payload = ConversationRenamePayload.model_validate(console_ns.payload or {})
-
         try:
             session = db.session()
             conversation = ConversationService.rename(
-                app_model, conversation_id, current_user, payload.name, payload.auto_generate, session=session
+                app_model, conversation_id, current_user, req_data.name, req_data.auto_generate, session=session
             )
             return (
                 TypeAdapter(SimpleConversation)

api/controllers/console/explore/installed_app.py

@@ -15,6 +15,7 @@ from controllers.console.explore.wraps import InstalledAppResource
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
+    model_validate,
     with_current_tenant_id,
     with_current_user,
 )
@@ -195,16 +196,15 @@ class InstalledAppsListApi(Resource):
     @console_ns.expect(console_ns.models[InstalledAppCreatePayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[SimpleMessageResponse.__name__])
     @with_current_tenant_id
-    def post(self, current_tenant_id: str):
-        payload = InstalledAppCreatePayload.model_validate(console_ns.payload or {})
-
+    @model_validate(InstalledAppCreatePayload)
+    def post(self, req_data: InstalledAppCreatePayload, current_tenant_id: str):
         recommended_app = db.session.scalar(
-            select(RecommendedApp).where(RecommendedApp.app_id == payload.app_id).limit(1)
+            select(RecommendedApp).where(RecommendedApp.app_id == req_data.app_id).limit(1)
         )
         if recommended_app is None:
             raise NotFound("Recommended app not found")
 
-        app = db.session.get(App, payload.app_id)
+        app = db.session.get(App, req_data.app_id)
 
         if app is None:
             raise NotFound("App entity not found")
@@ -214,7 +214,7 @@ class InstalledAppsListApi(Resource):
 
         installed_app = db.session.scalar(
             select(InstalledApp)
-            .where(and_(InstalledApp.app_id == payload.app_id, InstalledApp.tenant_id == current_tenant_id))
+            .where(and_(InstalledApp.app_id == req_data.app_id, InstalledApp.tenant_id == current_tenant_id))
             .limit(1)
         )
 
@@ -223,7 +223,7 @@ class InstalledAppsListApi(Resource):
             recommended_app.install_count += 1
 
             new_installed_app = InstalledApp(
-                app_id=payload.app_id,
+                app_id=req_data.app_id,
                 tenant_id=current_tenant_id,
                 app_owner_tenant_id=app.tenant_id,
                 is_pinned=False,
@@ -281,12 +281,11 @@ class InstalledAppApi(InstalledAppResource):
 
     @console_ns.response(200, "Success", console_ns.models[SimpleResultMessageResponse.__name__])
     @console_ns.expect(console_ns.models[InstalledAppUpdatePayload.__name__])
-    def patch(self, installed_app: InstalledApp):
-        payload = InstalledAppUpdatePayload.model_validate(console_ns.payload or {})
-
+    @model_validate(InstalledAppUpdatePayload)
+    def patch(self, req_data: InstalledAppUpdatePayload, installed_app: InstalledApp):
         commit_args = False
-        if payload.is_pinned is not None:
-            installed_app.is_pinned = payload.is_pinned
+        if req_data.is_pinned is not None:
+            installed_app.is_pinned = req_data.is_pinned
             commit_args = True
 
         if commit_args:

api/controllers/console/explore/message.py

@@ -24,7 +24,7 @@ from controllers.console.explore.error import (
     NotCompletionAppError,
 )
 from controllers.console.explore.wraps import InstalledAppResource
-from controllers.console.wraps import with_current_user
+from controllers.console.wraps import model_validate, with_current_user
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from extensions.ext_database import db
@@ -119,22 +119,23 @@ class MessageFeedbackApi(InstalledAppResource):
     @console_ns.expect(console_ns.models[MessageFeedbackPayload.__name__])
     @console_ns.response(200, "Feedback submitted successfully", console_ns.models[ResultResponse.__name__])
     @with_current_user
-    def post(self, current_user: Account, installed_app: InstalledApp, message_id: UUID):
+    @model_validate(MessageFeedbackPayload)
+    def post(
+        self, req_data: MessageFeedbackPayload, current_user: Account, installed_app: InstalledApp, message_id: UUID
+    ):
         app_model = installed_app.app_with_session(session=db.session())
         if app_model is None:
             raise AppUnavailableError()
 
         message_id_str = str(message_id)
 
-        payload = MessageFeedbackPayload.model_validate(console_ns.payload or {})
-
         try:
             MessageService.create_feedback(
                 app_model=app_model,
                 message_id=message_id_str,
                 user=current_user,
-                rating=FeedbackRating(payload.rating) if payload.rating else None,
-                content=payload.content,
+                rating=FeedbackRating(req_data.rating) if req_data.rating else None,
+                content=req_data.content,
                 session=db.session(),
             )
         except MessageNotExistsError:

api/controllers/console/explore/saved_message.py

@@ -10,7 +10,7 @@ from controllers.console import console_ns
 from controllers.console.app.error import AppUnavailableError
 from controllers.console.explore.error import NotCompletionAppError
 from controllers.console.explore.wraps import InstalledAppResource
-from controllers.console.wraps import with_current_user
+from controllers.console.wraps import model_validate, with_current_user
 from extensions.ext_database import db
 from fields.conversation_fields import MessageResponseSource, ResultResponse
 from fields.message_fields import SavedMessageInfiniteScrollPagination, SavedMessageItem
@@ -55,17 +55,16 @@ class SavedMessageListApi(InstalledAppResource):
     @console_ns.expect(console_ns.models[SavedMessageCreatePayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[ResultResponse.__name__])
     @with_current_user
-    def post(self, current_user: Account, installed_app: InstalledApp):
+    @model_validate(SavedMessageCreatePayload)
+    def post(self, req_data: SavedMessageCreatePayload, current_user: Account, installed_app: InstalledApp):
         app_model = installed_app.app_with_session(session=db.session())
         if app_model is None:
             raise AppUnavailableError()
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        payload = SavedMessageCreatePayload.model_validate(console_ns.payload or {})
-
         try:
-            SavedMessageService.save(app_model, current_user, str(payload.message_id), session=db.session())
+            SavedMessageService.save(app_model, current_user, str(req_data.message_id), session=db.session())
         except MessageNotExistsError:
             raise NotFound("Message Not Exists.")
 

api/controllers/console/explore/workflow.py

@@ -15,7 +15,7 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import with_session
 from controllers.console.explore.error import NotWorkflowAppError
 from controllers.console.explore.wraps import InstalledAppResource
-from controllers.console.wraps import with_current_user
+from controllers.console.wraps import model_validate, with_current_user
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -47,7 +47,14 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
     @console_ns.response(200, "Success")
     @with_current_user
     @with_session
-    def post(self, session: Session, current_user: Account, installed_app: InstalledApp):
+    @model_validate(WorkflowRunPayload)
+    def post(
+        self,
+        req_data: WorkflowRunPayload,
+        session: Session,
+        current_user: Account,
+        installed_app: InstalledApp,
+    ):
         """
         Run workflow
         """
@@ -58,8 +65,7 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        payload = WorkflowRunPayload.model_validate(console_ns.payload or {})
-        args = payload.model_dump(exclude_none=True)
+        args = req_data.model_dump(exclude_none=True)
         try:
             response = AppGenerateService.generate(
                 session=session,

api/controllers/console/setup.py

@@ -2,15 +2,16 @@ from typing import Literal
 
 from flask import request
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
 
-from configs import dify_config
 from controllers.fastopenapi import console_router
-from enums.deployment_edition import DeploymentEdition
+from extensions.ext_application_services import application_services
 from libs.helper import EmailStr, extract_remote_ip
 from libs.password import valid_password
-from models.model import DifySetup, db
-from services.account_service import RegisterService, TenantService
+from services.setup_service import (
+    InitializationValidationRequiredError,
+    SetupAlreadyCompletedError,
+    SetupInput,
+)
 
 from .error import AlreadySetupError, NotInitValidateError
 from .init_validate import get_init_validate_status
@@ -53,14 +54,12 @@ def get_setup_status_api() -> SetupStatusResponse:
 
     Only bootstrap-safe status information should be returned by this endpoint.
     """
-    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
-        setup_status = get_setup_status()
-        if setup_status and not isinstance(setup_status, bool):
-            return SetupStatusResponse(step="finished", setup_at=setup_status.setup_at.isoformat())
-        if setup_status:
-            return SetupStatusResponse(step="finished")
+    setup_status = application_services().setup.get_status()
+    if not setup_status.completed:
         return SetupStatusResponse(step="not_started")
-    return SetupStatusResponse(step="finished")
+
+    setup_at = setup_status.setup_at.isoformat() if setup_status.setup_at is not None else None
+    return SetupStatusResponse(step="finished", setup_at=setup_at)
 
 
 @console_router.post(
@@ -77,33 +76,22 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
     Access is restricted by deployment mode (`SELF_HOSTED`), one-time setup guards,
     and init-password validation rather than user session authentication.
     """
-    if get_setup_status():
-        raise AlreadySetupError()
+    try:
+        application_services().setup.initialize(
+            SetupInput(
+                email=payload.email,
+                name=payload.name,
+                password=payload.password,
+                ip_address=extract_remote_ip(request),
+                language=payload.language,
+            ),
+            initialization_validated=get_init_validate_status(),
+        )
+    except SetupAlreadyCompletedError:
+        raise AlreadySetupError() from None
+    except InitializationValidationRequiredError:
+        raise NotInitValidateError() from None
 
-    tenant_count = TenantService.get_tenant_count(session=db.session())
-    if tenant_count > 0:
-        raise AlreadySetupError()
-
-    if not get_init_validate_status():
-        raise NotInitValidateError()
-
-    normalized_email = payload.email.lower()
-
-    RegisterService.setup(
-        email=normalized_email,
-        name=payload.name,
-        password=payload.password,
-        ip_address=extract_remote_ip(request),
-        language=payload.language,
-        session=db.session(),
-    )
     mark_setup_completed()
 
     return SetupResponse(result="success")
-
-
-def get_setup_status() -> DifySetup | bool | None:
-    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
-        return db.session.scalar(select(DifySetup).limit(1))
-
-    return True

api/controllers/console/spec.py

@@ -1,23 +1,18 @@
-import logging
 from collections.abc import Mapping
+from http import HTTPStatus
 from typing import Any
 
 from flask_restx import Resource
 from pydantic import Field, RootModel
 
 from controllers.common.schema import register_response_schema_models
-from controllers.console.wraps import (
-    account_initialization_required,
-    setup_required,
-)
-from core.schemas.schema_manager import SchemaManager
+from controllers.console.flask_admission import console_account_admission
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
-from libs.login import login_required
+from machinery.context import RequestContext
 
 from . import console_ns
 
-logger = logging.getLogger(__name__)
-
 
 class SchemaDefinitionItemResponse(ResponseModel):
     name: str
@@ -34,20 +29,13 @@ register_response_schema_models(console_ns, SchemaDefinitionItemResponse, Schema
 
 @console_ns.route("/spec/schema-definitions")
 class SpecSchemaDefinitionsApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[SchemaDefinitionsResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
+    @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[SchemaDefinitionsResponse.__name__])
+    @console_account_admission()
+    def get(self, _request_context: RequestContext):
         """
         Get system JSON Schema definitions specification
         Used for frontend component type mapping
         """
-        try:
-            schema_manager = SchemaManager()
-            schema_definitions = schema_manager.get_all_schema_definitions()
-            return schema_definitions, 200
-        except Exception:
-            logger.exception("Failed to get schema definitions from local registry")
-            # Return empty array as fallback
-            return [], 200
+        schema_definitions = application_services().schema_definitions.list()
+        response = SchemaDefinitionsResponse.model_validate(schema_definitions).model_dump(mode="json")
+        return response, HTTPStatus.OK

api/core/tools/utils/dataset_retriever/dataset_retriever_tool.py

@@ -123,7 +123,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
             for hit_callback in self.hit_callbacks:
                 hit_callback.return_retriever_resource_info(context_list)
 
-            return str("\n".join([item.page_content for item in results]))
+            return "\n".join([item.page_content for item in results])
         else:
             if metadata_condition and not document_ids_filter:
                 return ""
@@ -139,7 +139,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                     top_k=self.top_k,
                     document_ids_filter=document_ids_filter,
                 )
-                return str("\n".join([document.page_content for document in documents]))
+                return "\n".join([document.page_content for document in documents])
             else:
                 if self.top_k > 0:
                     # retrieval source
@@ -241,5 +241,5 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                     hit_callback.return_retriever_resource_info(retrieval_resource_list)
             if document_context_list:
                 document_context_list = sorted(document_context_list, key=lambda x: x.score or 0.0, reverse=True)
-                return str("\n".join([document_context.content for document_context in document_context_list]))
+                return "\n".join([document_context.content for document_context in document_context_list])
             return ""

api/core/workflow/nodes/agent_v2/runtime_request_builder.py

@@ -836,7 +836,12 @@ def _knowledge_metadata_filtering_config(
     return DifyKnowledgeMetadataFilteringConfig(
         mode=metadata_filtering.mode,
         model_config=_knowledge_model_config(metadata_filtering.metadata_model_config),
-        conditions=cast(Any, metadata_filtering.conditions.model_dump(mode="json"))
+        conditions=cast(
+            Any,
+            metadata_filtering.conditions.model_dump(
+                mode="json", exclude={"conditions": {"__all__": {"id", "metadata_id"}}}
+            ),
+        )
         if metadata_filtering.conditions is not None
         else None,
     )

api/extensions/ext_application_services.py

@@ -6,13 +6,21 @@ from typing import cast
 from flask import Flask, current_app
 from sqlalchemy.orm import Session, sessionmaker
 
+from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
 from core.db.session_factory import get_session_maker
+from core.schemas.schema_manager import SchemaManager
+from enums.deployment_edition import DeploymentEdition
+from extensions.ext_redis import RedisClientWrapper, redis_client
+from repositories.installation_state_repository import InstallationStateRepository
 from repositories.workspace_member_query_repository import WorkspaceMemberQueryRepository
 from repositories.workspace_query_repository import WorkspaceQueryRepository
 from services.feature_query_service import FeatureQueryService
 from services.feature_service import FeatureService
 from services.feature_service_gateway import FeatureServiceGateway
+from services.schema_definition_service import SchemaDefinitionService
+from services.setup_adapters import RedisSetupLock, RegisterServiceAccountProvisioner
+from services.setup_service import SetupService
 from services.workspace_member_query_service import WorkspaceMemberQueryService
 from services.workspace_member_role_resolver import DeploymentWorkspaceMemberRoleResolver
 from services.workspace_plan_gateway import DeploymentWorkspacePlanGateway
@@ -23,6 +31,8 @@ _EXTENSION_KEY = "application_services"
 
 @dataclass(frozen=True, slots=True)
 class ApplicationServices:
+    schema_definitions: SchemaDefinitionService
+    setup: SetupService
     feature_queries: FeatureQueryService
     workspace_queries: WorkspaceQueryService
     workspace_member_queries: WorkspaceMemberQueryService
@@ -31,8 +41,18 @@ class ApplicationServices:
 def build_application_services(
     *,
     database_client: sessionmaker[Session],
+    deployment_edition: DeploymentEdition,
+    redis: RedisClientWrapper,
 ) -> ApplicationServices:
+    installation_state = InstallationStateRepository(client=database_client)
     return ApplicationServices(
+        schema_definitions=SchemaDefinitionService(source_factory=SchemaManager),
+        setup=SetupService(
+            state=installation_state,
+            accounts=RegisterServiceAccountProvisioner(client=database_client),
+            lock=RedisSetupLock(client=redis),
+            setup_required=deployment_edition != DeploymentEdition.CLOUD,
+        ),
         feature_queries=FeatureQueryService(
             features=FeatureServiceGateway(),
             trial_models=FeatureService.get_trial_models(),
@@ -56,6 +76,8 @@ def build_application_services(
 def init_app(app: Flask) -> None:
     app.extensions[_EXTENSION_KEY] = build_application_services(
         database_client=get_session_maker(),
+        deployment_edition=dify_config.DEPLOYMENT_EDITION,
+        redis=redis_client,
     )
 
 

api/models/agent_config_entities.py

@@ -420,8 +420,19 @@ class AgentKnowledgeRetrievalConfig(BaseModel):
 
 
 class AgentKnowledgeMetadataCondition(BaseModel):
+    """One manual metadata filter clause.
+
+    ``id`` and ``metadata_id`` are UI-only bookkeeping the composer sends on
+    every save (a stable row key and a reference to the selected metadata
+    field). They are persisted here for round-tripping the composer's draft
+    state but are stripped before building the Agent runtime request, whose
+    DTO only accepts ``name``/``comparison_operator``/``value``.
+    """
+
     model_config = ConfigDict(extra="forbid")
 
+    id: str | None = None
+    metadata_id: str | None = None
     name: str = Field(min_length=1, max_length=255)
     comparison_operator: SupportedComparisonOperator
     value: ConditionValue = None

api/openapi/markdown/console-openapi.md

@@ -14302,9 +14302,19 @@ the current roster/workflow APIs scoped to Dify Agent.
 
 #### AgentKnowledgeMetadataCondition
 
+One manual metadata filter clause.
+
+``id`` and ``metadata_id`` are UI-only bookkeeping the composer sends on
+every save (a stable row key and a reference to the selected metadata
+field). They are persisted here for round-tripping the composer's draft
+state but are stripped before building the Agent runtime request, whose
+DTO only accepts ``name``/``comparison_operator``/``value``.
+
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | comparison_operator | string, <br>**Available values:** "<", "=", ">", "after", "before", "contains", "empty", "end with", "in", "is", "is not", "not contains", "not empty", "not in", "start with", "≠", "≤", "≥" | *Enum:* `"<"`, `"="`, `">"`, `"after"`, `"before"`, `"contains"`, `"empty"`, `"end with"`, `"in"`, `"is"`, `"is not"`, `"not contains"`, `"not empty"`, `"not in"`, `"start with"`, `"≠"`, `"≤"`, `"≥"` | Yes |
+| id | string |  | No |
+| metadata_id | string |  | No |
 | name | string |  | Yes |
 | value | string<br>[ string ]<br>number |  | No |
 

api/repositories/installation_state_repository.py

@@ -0,0 +1,27 @@
+"""Persistence adapter for installation setup and tenant-existence state."""
+
+from datetime import datetime
+
+from sqlalchemy import exists, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from models.account import Tenant
+from models.model import DifySetup
+
+
+class InstallationStateRepository:
+    """Read persistent state shared by installation bootstrap use cases."""
+
+    def __init__(self, client: sessionmaker[Session]) -> None:
+        self._client = client
+
+    def get_setup_at(self) -> datetime | None:
+        with self._client() as session:
+            return session.scalar(select(DifySetup.setup_at).limit(1))
+
+    def is_setup(self) -> bool:
+        return self.get_setup_at() is not None
+
+    def has_tenants(self) -> bool:
+        with self._client() as session:
+            return session.scalar(select(exists().select_from(Tenant))) is True

api/services/schema_definition_service.py

@@ -0,0 +1,24 @@
+"""Application service for querying Console schema definitions."""
+
+import logging
+from collections.abc import Callable, Mapping
+from typing import Any, Protocol
+
+logger = logging.getLogger(__name__)
+
+
+class SchemaDefinitionSource(Protocol):
+    def get_all_schema_definitions(self, version: str = "v1") -> list[Mapping[str, Any]]: ...
+
+
+class SchemaDefinitionService:
+    def __init__(self, *, source_factory: Callable[[], SchemaDefinitionSource]) -> None:
+        self._source_factory = source_factory
+
+    def list(self) -> tuple[Mapping[str, Any], ...]:
+        try:
+            source = self._source_factory()
+            return tuple(source.get_all_schema_definitions())
+        except Exception:
+            logger.exception("Failed to get schema definitions from local registry")
+            return ()

api/services/setup_adapters.py

@@ -0,0 +1,43 @@
+"""Infrastructure adapters for the first-time setup application service."""
+
+from contextlib import AbstractContextManager
+from typing import override
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from extensions.ext_redis import RedisClientWrapper
+from services.account_service import RegisterService
+from services.setup_service import SetupAccountProvisioner, SetupInput, SetupLock
+
+_SETUP_LOCK_KEY = "setup:initialize"
+_SETUP_LOCK_TIMEOUT_SECONDS = 300
+
+
+class RegisterServiceAccountProvisioner(SetupAccountProvisioner):
+    def __init__(self, client: sessionmaker[Session]) -> None:
+        self._client = client
+
+    @override
+    def provision(self, setup: SetupInput) -> None:
+        with self._client() as session:
+            RegisterService.setup(
+                email=setup.email,
+                name=setup.name,
+                password=setup.password,
+                ip_address=setup.ip_address,
+                language=setup.language,
+                session=session,
+            )
+
+
+class RedisSetupLock(SetupLock):
+    def __init__(self, *, client: RedisClientWrapper) -> None:
+        self._client = client
+
+    @override
+    def acquire(self) -> AbstractContextManager[None]:
+        return self._client.lock(
+            _SETUP_LOCK_KEY,
+            timeout=_SETUP_LOCK_TIMEOUT_SECONDS,
+            blocking_timeout=_SETUP_LOCK_TIMEOUT_SECONDS,
+        )

api/services/setup_service.py

@@ -0,0 +1,83 @@
+"""Application service for first-time Dify setup."""
+
+from contextlib import AbstractContextManager
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Protocol
+
+
+@dataclass(frozen=True, slots=True)
+class SetupInput:
+    email: str
+    name: str
+    password: str
+    ip_address: str
+    language: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class SetupStatus:
+    completed: bool
+    setup_at: datetime | None = None
+
+
+class SetupState(Protocol):
+    def get_setup_at(self) -> datetime | None: ...
+
+    def has_tenants(self) -> bool: ...
+
+
+class SetupAccountProvisioner(Protocol):
+    def provision(self, setup: SetupInput) -> None: ...
+
+
+class SetupLock(Protocol):
+    def acquire(self) -> AbstractContextManager[None]: ...
+
+
+class SetupAlreadyCompletedError(Exception):
+    """Raised when setup has already created persistent installation state."""
+
+
+class InitializationValidationRequiredError(Exception):
+    """Raised when initialization-password validation has not completed."""
+
+
+class SetupService:
+    def __init__(
+        self,
+        *,
+        state: SetupState,
+        accounts: SetupAccountProvisioner,
+        lock: SetupLock,
+        setup_required: bool,
+    ) -> None:
+        self._state = state
+        self._accounts = accounts
+        self._lock = lock
+        self._setup_required = setup_required
+
+    def get_status(self) -> SetupStatus:
+        if not self._setup_required:
+            return SetupStatus(completed=True)
+
+        setup_at = self._state.get_setup_at()
+        return SetupStatus(completed=setup_at is not None, setup_at=setup_at)
+
+    def initialize(self, setup: SetupInput, *, initialization_validated: bool) -> None:
+        with self._lock.acquire():
+            if self._state.get_setup_at() is not None or self._state.has_tenants():
+                raise SetupAlreadyCompletedError
+
+            if not initialization_validated:
+                raise InitializationValidationRequiredError
+
+            self._accounts.provision(
+                SetupInput(
+                    email=setup.email.lower(),
+                    name=setup.name,
+                    password=setup.password,
+                    ip_address=setup.ip_address,
+                    language=setup.language,
+                )
+            )

api/tests/test_containers_integration_tests/controllers/console/explore/test_conversation.py

@@ -198,7 +198,13 @@ class TestConversationRenameApi:
                 return_value=conversation,
             ),
         ):
-            result = method(api, user, chat_app, "cid")
+            result = method(
+                api,
+                conversation_module.ConversationRenamePayload.model_validate({"name": "new"}),
+                user,
+                chat_app,
+                "cid",
+            )
 
         assert result["id"] == "cid"
 
@@ -215,7 +221,13 @@ class TestConversationRenameApi:
             ),
         ):
             with pytest.raises(NotFound):
-                method(api, user, chat_app, "cid")
+                method(
+                    api,
+                    conversation_module.ConversationRenamePayload.model_validate({"name": "new"}),
+                    user,
+                    chat_app,
+                    "cid",
+                )
 
 
 class TestConversationPinApi:

api/tests/test_containers_integration_tests/controllers/console/test_setup.py

@@ -0,0 +1,159 @@
+from collections.abc import Iterator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from threading import Event, Lock
+from unittest.mock import MagicMock, patch
+
+import pytest
+from faker import Faker
+from flask import Flask
+from flask.testing import FlaskClient
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from models.account import Account, Tenant, TenantAccountJoin
+from models.model import DifySetup
+from services.account_service import RegisterService
+from services.setup_adapters import RedisSetupLock
+from tests.test_containers_integration_tests.helpers import generate_valid_password
+
+
+@pytest.fixture
+def setup_dependencies() -> Iterator[MagicMock]:
+    with (
+        patch("services.account_service.FeatureService") as feature_service,
+        patch("services.account_service.BillingService") as billing_service,
+        patch("services.account_service.CommunityTelemetryService.report_install") as report_install,
+    ):
+        feature_service.get_system_features.return_value.is_allow_register = True
+        feature_service.get_license.return_value.seats.is_available.return_value = True
+        feature_service.get_license.return_value.workspaces.is_available.return_value = True
+        feature_service.is_workspace_creation_allowed.return_value = True
+        billing_service.is_email_in_freeze.return_value = False
+        yield report_install
+
+
+def _setup_payload(*, email: str, password: str) -> dict[str, str]:
+    return {
+        "email": email,
+        "name": "Admin",
+        "password": password,
+        "language": "en-US",
+    }
+
+
+def test_setup_endpoint_persists_bootstrap_state_and_rejects_repeat(
+    db_session_with_containers: Session,
+    test_client_with_containers: FlaskClient,
+    monkeypatch: pytest.MonkeyPatch,
+    setup_dependencies: MagicMock,
+) -> None:
+    monkeypatch.delenv("INIT_PASSWORD", raising=False)
+    password = generate_valid_password(Faker())
+
+    response = test_client_with_containers.post(
+        "/console/api/setup",
+        json=_setup_payload(email="admin@example.com", password=password),
+        headers={"CF-Connecting-IP": "203.0.113.7"},
+    )
+
+    assert response.status_code == 201
+    assert response.get_json() == {"result": "success"}
+    setup_dependencies.assert_called_once()
+
+    repeated_response = test_client_with_containers.post(
+        "/console/api/setup",
+        json=_setup_payload(email="other@example.com", password=password),
+    )
+
+    assert repeated_response.status_code == 403
+
+    db_session_with_containers.expire_all()
+    assert db_session_with_containers.scalar(select(func.count()).select_from(DifySetup)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(Account)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(Tenant)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(TenantAccountJoin)) == 1
+
+    account = db_session_with_containers.scalar(select(Account))
+    assert account is not None
+    assert account.email == "admin@example.com"
+    assert account.last_login_ip == "203.0.113.7"
+
+
+def test_concurrent_setup_requests_create_only_one_bootstrap_identity(
+    flask_app_with_containers: Flask,
+    db_session_with_containers: Session,
+    monkeypatch: pytest.MonkeyPatch,
+    setup_dependencies: MagicMock,
+) -> None:
+    monkeypatch.delenv("INIT_PASSWORD", raising=False)
+    password = generate_valid_password(Faker())
+    provision_started = Event()
+    allow_provision_to_finish = Event()
+    second_lock_attempted = Event()
+    attempt_guard = Lock()
+    lock_attempts = 0
+    original_setup = RegisterService.setup
+    original_acquire = RedisSetupLock.acquire
+
+    def blocking_setup(
+        email: str,
+        name: str,
+        password: str,
+        ip_address: str,
+        language: str | None,
+        *,
+        session: Session,
+    ) -> None:
+        if not provision_started.is_set():
+            provision_started.set()
+            assert allow_provision_to_finish.wait(timeout=10)
+        original_setup(
+            email=email,
+            name=name,
+            password=password,
+            ip_address=ip_address,
+            language=language,
+            session=session,
+        )
+
+    def tracked_acquire(self: RedisSetupLock) -> AbstractContextManager[None]:
+        nonlocal lock_attempts
+        with attempt_guard:
+            lock_attempts += 1
+            if lock_attempts == 2:
+                second_lock_attempted.set()
+        return original_acquire(self)
+
+    def post_setup(email: str) -> int:
+        with flask_app_with_containers.test_client() as client:
+            response = client.post(
+                "/console/api/setup",
+                json=_setup_payload(email=email, password=password),
+            )
+            return response.status_code
+
+    with (
+        patch.object(RegisterService, "setup", side_effect=blocking_setup),
+        patch.object(RedisSetupLock, "acquire", tracked_acquire),
+        ThreadPoolExecutor(max_workers=2) as executor,
+    ):
+        first_request = executor.submit(post_setup, "admin-1@example.com")
+        try:
+            assert provision_started.wait(timeout=10)
+            second_request = executor.submit(post_setup, "admin-2@example.com")
+            assert second_lock_attempted.wait(timeout=10)
+            assert not second_request.done()
+        finally:
+            allow_provision_to_finish.set()
+
+        results = [first_request.result(timeout=30), second_request.result(timeout=30)]
+
+    assert sorted(results) == [201, 403]
+    setup_dependencies.assert_called_once()
+
+    db_session_with_containers.expire_all()
+    assert db_session_with_containers.scalar(select(func.count()).select_from(DifySetup)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(Account)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(Tenant)) == 1
+    assert db_session_with_containers.scalar(select(func.count()).select_from(TenantAccountJoin)) == 1

api/tests/test_containers_integration_tests/controllers/console/test_spec.py

@@ -0,0 +1,37 @@
+from flask.testing import FlaskClient
+from sqlalchemy.orm import Session
+
+from tests.test_containers_integration_tests.controllers.console.helpers import (
+    authenticate_console_client,
+    create_console_account_and_tenant,
+)
+
+
+def test_schema_definitions_endpoint_uses_admission_and_builtin_registry(
+    db_session_with_containers: Session,
+    test_client_with_containers: FlaskClient,
+) -> None:
+    account, _ = create_console_account_and_tenant(db_session_with_containers)
+    headers = authenticate_console_client(test_client_with_containers, account)
+
+    response = test_client_with_containers.get(
+        "/console/api/spec/schema-definitions",
+        headers=headers,
+    )
+
+    assert response.status_code == 200
+    definitions = response.get_json()
+    assert isinstance(definitions, list)
+    assert definitions
+    assert all({"name", "label", "schema"} <= definition.keys() for definition in definitions)
+
+
+def test_schema_definitions_endpoint_rejects_unauthenticated_request(
+    db_session_with_containers: Session,
+    test_client_with_containers: FlaskClient,
+) -> None:
+    create_console_account_and_tenant(db_session_with_containers)
+
+    response = test_client_with_containers.get("/console/api/spec/schema-definitions")
+
+    assert response.status_code == 401

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

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 from inspect import unwrap
 from types import SimpleNamespace
-from unittest.mock import MagicMock, PropertyMock, patch
+from unittest.mock import MagicMock, patch
 
 from flask import Flask
 from sqlalchemy import event
@@ -28,7 +28,6 @@ from controllers.console.app.agent_config_inspector import (
     AgentConfigSkillInspectByAgentApi,
     AgentConfigSkillsApi,
     AgentConfigSkillUploadByAgentApi,
-    console_ns,
 )
 from services.agent_config_service import AgentConfigServiceError
 
@@ -212,13 +211,10 @@ def test_skill_upload_by_agent_delegates_after_version_resolution():
 
 def test_file_upload_by_agent_delegates_to_service_owned_upload_lookup():
     raw = _raw(AgentConfigFilesByAgentApi.post)
-    with app.test_request_context("/?draft_type=debug_build"):
+    with app.test_request_context("/?draft_type=debug_build", json={"upload_file_id": "upload-1"}):
         with (
             patch(f"{_MOD}.resolve_agent_runtime_app_model", return_value=_APP),
             patch(f"{_MOD}.AgentComposerService") as composer,
-            patch.object(
-                type(console_ns), "payload", new_callable=PropertyMock, return_value={"upload_file_id": "upload-1"}
-            ),
             patch(f"{_MOD}.AgentConfigService") as config_service,
         ):
             composer.load_agent_app_build_draft.return_value = {"draft": {"id": "build-draft-1"}}
@@ -226,7 +222,14 @@ def test_file_upload_by_agent_delegates_to_service_owned_upload_lookup():
                 "file": {"id": "guide.txt", "name": "guide.txt", "file_id": "upload-1"},
                 "config_version": {"id": "build-draft-1", "kind": "build_draft", "writable": True},
             }
-            body, status = raw(AgentConfigFilesByAgentApi(), MagicMock(), "tenant-1", _USER, "agent-1")
+            body, status = raw(
+                AgentConfigFilesByAgentApi(),
+                inspector.AgentConfigFileUploadPayload(upload_file_id="upload-1"),
+                MagicMock(),
+                "tenant-1",
+                _USER,
+                "agent-1",
+            )
     assert status == 201
     assert body["file"]["name"] == "guide.txt"
     assert config_service.return_value.push_file_for_console.call_args.kwargs["upload_file_id"] == "upload-1"

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

@@ -163,12 +163,10 @@ class TestConsoleAnnotationRefBoundaries:
         api = annotation_module.AnnotationUpdateDeleteApi()
         handler = unwrap(api.post)
         update_mock = Mock(return_value=_annotation_model())
-        payload = {"question": "updated"}
         _persist_app(sqlite_session)
 
         with (
-            app.test_request_context("/annotations/ann-1", method="POST", json=payload),
-            patch.object(type(annotation_module.console_ns), "payload", payload),
+            app.test_request_context("/annotations/ann-1", method="POST", json={"question": "updated"}),
             patch.object(
                 annotation_module,
                 "current_account_with_tenant",
@@ -176,7 +174,13 @@ class TestConsoleAnnotationRefBoundaries:
             ),
             patch.object(annotation_module.AppAnnotationService, "update_app_annotation_directly", update_mock),
         ):
-            response = handler(api, sqlite_session, "app-1", "ann-1")
+            response = handler(
+                api,
+                annotation_module.UpdateAnnotationPayload(question="updated"),
+                sqlite_session,
+                "app-1",
+                "ann-1",
+            )
 
         assert response["question"] == "q"
         update_mock.assert_called_once()

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

@@ -301,7 +301,7 @@ class TestOpsTraceEndpoints:
         )
 
         with app.test_request_context("/?tracing_provider=langfuse"):
-            result = method(api, app_model=MagicMock(id="app-1"))
+            result = method(api, TraceProviderQuery(tracing_provider="langfuse"), MagicMock(id="app-1"))
 
         assert result == {"has_not_configured": True}
 
@@ -320,7 +320,11 @@ class TestOpsTraceEndpoints:
             json={"tracing_provider": "langfuse", "tracing_config": {"api_key": "k"}},
         ):
             with pytest.raises(BadRequest):
-                method(api, app_model=MagicMock(id="app-1"))
+                method(
+                    api,
+                    TraceConfigPayload(tracing_provider="langfuse", tracing_config={"api_key": "k"}),
+                    MagicMock(id="app-1"),
+                )
 
     def test_trace_app_config_delete_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatch):
         api = ops_trace_module.TraceAppConfigApi()
@@ -334,7 +338,7 @@ class TestOpsTraceEndpoints:
 
         with app.test_request_context("/?tracing_provider=langfuse"):
             with pytest.raises(BadRequest):
-                method(api, app_model=MagicMock(id="app-1"))
+                method(api, TraceProviderQuery(tracing_provider="langfuse"), MagicMock(id="app-1"))
 
 
 class TestSiteEndpoints:

api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py

@@ -1,40 +1,51 @@
-"""RAG pipeline workflow controller serialization tests.
-
-Handlers that own transactions run against real SQLite sessions so response
-DTOs must be materialized before those transaction contexts close.
-"""
+"""Unit coverage for RAG workflow controllers using real models and disposable SQLite state."""
 
 from __future__ import annotations
 
+import json
+from collections.abc import Iterator
 from datetime import datetime
 from inspect import unwrap as unwrap_all
-from types import SimpleNamespace
-from unittest.mock import PropertyMock, patch
+from unittest.mock import MagicMock, patch
+from uuid import UUID
 
 import pytest
 from flask import Flask
-from sqlalchemy.engine import Engine
+from sqlalchemy import Engine
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from controllers.console.datasets.rag_pipeline import rag_pipeline_workflow as module
+from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from models.account import Account, TenantAccountRole
 from models.dataset import Pipeline
+from models.engine import db
+from models.tools import WorkflowToolProvider
+from models.workflow import Workflow, WorkflowType
+from services.errors.llm import InvokeRateLimitError
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+DEFAULT_WORKFLOW_TENANT_ID = "00000000-0000-0000-0000-000000000001"
+DEFAULT_WORKFLOW_APP_ID = "00000000-0000-0000-0000-000000000002"
+DEFAULT_WORKFLOW_CREATED_BY = "00000000-0000-0000-0000-000000000003"
+DEFAULT_WORKFLOW_ID = "00000000-0000-0000-0000-000000000004"
 
 
-def _make_workflow(**overrides):
-    workflow = SimpleNamespace(
-        id="workflow-1",
-        graph_dict={"nodes": [], "edges": []},
-        features_dict={"file_upload": {"enabled": False}},
-        unique_hash="hash-1",
-        version="1",
+def _make_workflow(**overrides: object) -> Workflow:
+    workflow = Workflow(
+        id=DEFAULT_WORKFLOW_ID,
+        tenant_id=DEFAULT_WORKFLOW_TENANT_ID,
+        app_id=DEFAULT_WORKFLOW_APP_ID,
+        type=WorkflowType.WORKFLOW,
+        version=Workflow.VERSION_DRAFT,
         marked_name="Release 1",
         marked_comment="Initial release",
-        created_by_account=SimpleNamespace(id="user-1", name="Alice", email="alice@example.com"),
+        graph=json.dumps({"nodes": [], "edges": []}),
+        features=json.dumps({"file_upload": {"enabled": False}}),
+        created_by=DEFAULT_WORKFLOW_CREATED_BY,
         created_at=datetime(2024, 1, 1, 12, 0, 0),
-        updated_by_account=None,
+        updated_by=None,
         updated_at=datetime(2024, 1, 1, 12, 1, 0),
-        tool_published=False,
         environment_variables=[],
         conversation_variables=[],
         rag_pipeline_variables=[],
@@ -46,137 +57,130 @@ def _make_workflow(**overrides):
 
 def _account() -> Account:
     account = Account(name="Alice", email="alice@example.com")
-    account.id = "user-1"
+    account.id = DEFAULT_WORKFLOW_CREATED_BY
     account.role = TenantAccountRole.EDITOR
     return account
 
 
 def _pipeline() -> Pipeline:
-    pipeline = Pipeline(tenant_id="tenant-1", name="Pipeline", description="desc")
-    pipeline.id = "pipeline-1"
+    pipeline = Pipeline(tenant_id=DEFAULT_WORKFLOW_TENANT_ID, name="Pipeline", description="desc")
+    pipeline.id = DEFAULT_WORKFLOW_APP_ID
     return pipeline
 
 
-def test_draft_rag_pipeline_workflow_get_serializes_response_model(monkeypatch: pytest.MonkeyPatch) -> None:
+def _persist_workflow(workflow: Workflow) -> None:
+    db.session.add(workflow)
+    db.session.commit()
+    db.session.expunge(workflow)
+
+
+@pytest.fixture
+def database_app() -> Iterator[Flask]:
+    app = Flask(__name__)
+    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
+    db.init_app(app)
+
+    with app.app_context():
+        Account.__table__.create(db.engine)
+        WorkflowToolProvider.__table__.create(db.engine)
+        Workflow.__table__.create(db.engine)
+        db.session.add(_account())
+        db.session.commit()
+
+        try:
+            yield app
+        finally:
+            db.session.remove()
+
+
+@pytest.mark.usefixtures("database_app")
+def test_draft_rag_pipeline_workflow_get_serializes_response_model() -> None:
     workflow = _make_workflow()
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(get_draft_workflow=lambda **_kwargs: workflow),
-    )
+    expected_hash = workflow.unique_hash
+    _persist_workflow(workflow)
 
     api = module.DraftRagPipelineApi()
     handler = unwrap_all(api.get)
 
     response = handler(api, _pipeline())
 
-    assert response["id"] == "workflow-1"
+    assert response["id"] == DEFAULT_WORKFLOW_ID
     assert response["graph"] == {"nodes": [], "edges": []}
     assert response["features"] == {"file_upload": {"enabled": False}}
-    assert response["hash"] == "hash-1"
-    assert response["created_by"] == {"id": "user-1", "name": "Alice", "email": "alice@example.com"}
+    assert response["hash"] == expected_hash
+    assert response["created_by"] == {
+        "id": DEFAULT_WORKFLOW_CREATED_BY,
+        "name": "Alice",
+        "email": "alice@example.com",
+    }
     assert response["updated_by"] is None
     assert response["created_at"] == int(datetime(2024, 1, 1, 12, 0, 0).timestamp())
     assert response["updated_at"] == int(datetime(2024, 1, 1, 12, 1, 0).timestamp())
 
 
 def test_published_rag_pipeline_workflows_serialize_items_before_session_closes(
-    app, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine
+    database_app: Flask,
 ) -> None:
     api = module.PublishedAllRagPipelineApi()
     handler = unwrap_all(api.get)
-    session_state: dict[str, Session] = {}
+    workflow = _make_workflow(version="1")
+    _persist_workflow(workflow)
+    pipeline = _pipeline()
+    pipeline.workflow_id = DEFAULT_WORKFLOW_ID
 
-    base_workflow = _make_workflow()
+    with database_app.test_request_context(
+        "/rag/pipelines/pipeline-1/workflows",
+        method="GET",
+        query_string={"page": 1, "limit": 10, "user_id": "", "named_only": "false"},
+    ):
+        response = handler(api, _account(), pipeline=pipeline)
 
-    class _Workflow:
-        def __getattr__(self, name: str):
-            assert session_state["session"].in_transaction() is True
-            return getattr(base_workflow, name)
-
-    def _get_all_published_workflow(**kwargs):
-        session_state["session"] = kwargs["session"]
-        return [_Workflow()], False
-
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(get_all_published_workflow=_get_all_published_workflow),
-    )
-
-    with Session(sqlite_engine) as request_session:
-        monkeypatch.setattr(module, "db", SimpleNamespace(engine=sqlite_engine, session=lambda: request_session))
-        with app.test_request_context(
-            "/rag/pipelines/pipeline-1/workflows",
-            method="GET",
-            query_string={"page": 1, "limit": 10, "user_id": "", "named_only": "false"},
-        ):
-            response = handler(api, _account(), pipeline=_pipeline())
-
-    assert session_state["session"].in_transaction() is False
-    assert response["items"][0]["id"] == "workflow-1"
+    assert response["items"][0]["id"] == DEFAULT_WORKFLOW_ID
     assert response["page"] == 1
     assert response["limit"] == 10
     assert response["has_more"] is False
 
 
 def test_rag_pipeline_workflow_patch_serializes_response_model(
-    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine
+    database_app: Flask,
 ) -> None:
     workflow = _make_workflow(marked_name="Updated release")
-    captured_session: dict[str, Session] = {}
-
-    def _update_workflow(**kwargs):
-        captured_session["session"] = kwargs["session"]
-        assert kwargs["session"].in_transaction() is True
-        return workflow
-
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(update_workflow=_update_workflow),
-    )
+    expected_hash = workflow.unique_hash
+    _persist_workflow(workflow)
     payload: dict[str, object] = {"marked_name": "Updated release"}
 
     api = module.RagPipelineByIdApi()
     handler = unwrap_all(api.patch)
 
-    with Session(sqlite_engine) as request_session:
-        monkeypatch.setattr(module, "db", SimpleNamespace(engine=sqlite_engine, session=lambda: request_session))
-        with (
-            app.test_request_context("/rag/pipelines/pipeline-1/workflows/workflow-1", method="PATCH", json=payload),
-            patch.object(type(module.console_ns), "payload", new_callable=PropertyMock, return_value=payload),
-        ):
-            response = handler(
-                api,
-                _account(),
-                pipeline=_pipeline(),
-                workflow_id="workflow-1",
-            )
+    with database_app.test_request_context(
+        f"/rag/pipelines/{DEFAULT_WORKFLOW_APP_ID}/workflows/{DEFAULT_WORKFLOW_ID}", method="PATCH", json=payload
+    ):
+        response = handler(
+            api,
+            _account(),
+            pipeline=_pipeline(),
+            workflow_id=DEFAULT_WORKFLOW_ID,
+        )
 
-    assert captured_session["session"].in_transaction() is False
-    assert response["id"] == "workflow-1"
+    assert response["id"] == DEFAULT_WORKFLOW_ID
     assert response["marked_name"] == "Updated release"
-    assert response["hash"] == "hash-1"
+    assert response["hash"] == expected_hash
 
 
-def test_default_rag_pipeline_block_configs_serializes_root_response(monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.mark.usefixtures("database_app")
+def test_default_rag_pipeline_block_configs_serializes_root_response() -> None:
     block_configs = [{"type": "start", "config": {"title": "Start"}}]
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(get_default_block_configs=lambda: block_configs),
-    )
 
     api = module.DefaultRagPipelineBlockConfigsApi()
     handler = unwrap_all(api.get)
 
-    response = handler(api, _pipeline())
+    with patch.object(RagPipelineService, "get_default_block_configs", return_value=block_configs):
+        response = handler(api, _pipeline())
 
     assert response == block_configs
 
 
-def test_draft_rag_pipeline_second_step_parameters_serializes_variables(app, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_draft_rag_pipeline_second_step_parameters_serializes_variables(database_app: Flask) -> None:
     variables = [
         {
             "belong_to_node_id": "shared",
@@ -187,36 +191,114 @@ def test_draft_rag_pipeline_second_step_parameters_serializes_variables(app, mon
             "required": True,
         }
     ]
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(get_second_step_parameters=lambda **_kwargs: variables),
-    )
-
     api = module.DraftRagPipelineSecondStepApi()
     handler = unwrap_all(api.get)
 
-    with app.test_request_context("/?node_id=node-1"):
+    with (
+        database_app.test_request_context("/?node_id=node-1"),
+        patch.object(RagPipelineService, "get_second_step_parameters", return_value=variables),
+    ):
         response = handler(api, _pipeline())
 
     assert response["variables"] == variables
 
 
-def test_rag_pipeline_recommended_plugins_serializes_known_envelope(app, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_rag_pipeline_recommended_plugins_serializes_known_envelope(database_app: Flask) -> None:
     recommended_plugins = {
         "installed_recommended_plugins": [{"name": "Dify Extractor", "meta": {"version": "1.0.0"}}],
         "uninstalled_recommended_plugins": [{"plugin_id": "langgenius/notion_datasource"}],
     }
-    monkeypatch.setattr(
-        module,
-        "RagPipelineService",
-        lambda *_args, **_kwargs: SimpleNamespace(get_recommended_plugins=lambda *_args: recommended_plugins),
-    )
-
     api = module.RagPipelineRecommendedPluginApi()
     handler = unwrap_all(api.get)
 
-    with app.test_request_context("/?type=tool"):
-        response = handler(api, "tenant-1", _account())
+    with (
+        database_app.test_request_context("/?type=tool"),
+        patch.object(RagPipelineService, "get_recommended_plugins", return_value=recommended_plugins),
+    ):
+        response = handler(api, DEFAULT_WORKFLOW_TENANT_ID, _account())
 
     assert response == recommended_plugins
+
+
+def test_rag_pipeline_transform_rejects_read_only_member(app: Flask, sqlite_engine: Engine) -> None:
+    account = _account()
+    account.role = TenantAccountRole.NORMAL
+    api = module.RagPipelineTransformApi()
+    handler = unwrap_all(api.post)
+
+    with (
+        Session(sqlite_engine) as session,
+        app.test_request_context("/"),
+        pytest.raises(Forbidden),
+    ):
+        handler(api, session, account, UUID("44444444-4444-4444-4444-444444444444"))
+
+
+@pytest.mark.parametrize(
+    ("api_type", "payload"),
+    [
+        (
+            module.DraftRagPipelineRunApi,
+            {"inputs": {}, "datasource_type": "x", "datasource_info_list": [], "start_node_id": "node-1"},
+        ),
+        (
+            module.PublishedRagPipelineRunApi,
+            {
+                "inputs": {},
+                "datasource_type": "x",
+                "datasource_info_list": [],
+                "start_node_id": "node-1",
+                "response_mode": "blocking",
+            },
+        ),
+    ],
+)
+def test_rag_pipeline_run_uses_sqlite_session(
+    app: Flask,
+    sqlite_engine: Engine,
+    api_type: type,
+    payload: dict[str, object],
+) -> None:
+    api = api_type()
+    handler = unwrap_all(api.post)
+    pipeline = _pipeline()
+
+    with (
+        Session(sqlite_engine) as session,
+        app.test_request_context("/", json=payload),
+        patch.object(module, "load_rag_pipeline", return_value=pipeline) as load_pipeline,
+        patch.object(module.PipelineGenerateService, "generate", return_value=MagicMock()) as generate,
+        patch.object(module.helper, "compact_generate_response", return_value={"ok": True}),
+    ):
+        response = handler(api, session, _account(), pipeline.id)
+
+    assert response == {"ok": True}
+    load_pipeline.assert_called_once_with(session, pipeline.id)
+    assert generate.call_args.kwargs["session"] is session
+    assert session.get_bind() is sqlite_engine
+
+
+@pytest.mark.parametrize("api_type", [module.DraftRagPipelineRunApi, module.PublishedRagPipelineRunApi])
+def test_rag_pipeline_run_translates_rate_limit(
+    app: Flask,
+    sqlite_engine: Engine,
+    api_type: type,
+) -> None:
+    payload = {
+        "inputs": {},
+        "datasource_type": "x",
+        "datasource_info_list": [],
+        "start_node_id": "node-1",
+    }
+    api = api_type()
+    handler = unwrap_all(api.post)
+    pipeline = _pipeline()
+
+    with (
+        Session(sqlite_engine) as session,
+        app.test_request_context("/", json=payload),
+        patch.object(module, "load_rag_pipeline", return_value=pipeline),
+        patch.object(module.PipelineGenerateService, "generate", side_effect=InvokeRateLimitError("limit")),
+        pytest.raises(InvokeRateLimitHttpError),
+    ):
+        handler(api, session, _account(), pipeline.id)

api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow_apis.py

@@ -1,8 +1,10 @@
-"""Testcontainers integration tests for rag_pipeline_workflow controller endpoints."""
+"""Unit tests for rag_pipeline_workflow controller endpoints."""
 
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
+from dataclasses import dataclass
 from datetime import datetime
 from inspect import unwrap
 from typing import TypedDict, Unpack
@@ -11,19 +13,19 @@ from uuid import uuid4
 
 import pytest
 from flask import Flask
-from sqlalchemy.orm import Session
+from sqlalchemy import Engine
+from sqlalchemy.orm import Session, scoped_session, sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, HTTPException, NotFound
 
+import models.workflow as workflow_models
 import services
-from controllers.console import console_ns
 from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.datasets.rag_pipeline import rag_pipeline_workflow as workflow_controller
 from controllers.console.datasets.rag_pipeline.rag_pipeline_workflow import (
     DefaultRagPipelineBlockConfigApi,
     DraftRagPipelineApi,
-    DraftRagPipelineRunApi,
     PublishedAllRagPipelineApi,
     PublishedRagPipelineApi,
-    PublishedRagPipelineRunApi,
     RagPipelineByIdApi,
     RagPipelineDatasourceVariableApi,
     RagPipelineDraftNodeRunApi,
@@ -32,11 +34,9 @@ from controllers.console.datasets.rag_pipeline.rag_pipeline_workflow import (
     RagPipelineDraftWorkflowRestoreApi,
     RagPipelineRecommendedPluginApi,
     RagPipelineTaskStopApi,
-    RagPipelineTransformApi,
     RagPipelineWorkflowLastRunApi,
     RagPipelineWorkflowRunNodeExecutionListApi,
 )
-from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from graphon.enums import WorkflowNodeExecutionStatus
 from libs.datetime_utils import naive_utc_now
 from models.account import Account, TenantAccountRole
@@ -44,7 +44,6 @@ from models.dataset import Pipeline
 from models.enums import CreatorUserRole
 from models.workflow import Workflow, WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
-from services.errors.llm import InvokeRateLimitError
 
 DEFAULT_WORKFLOW_TENANT_ID = "00000000-0000-0000-0000-000000000001"
 DEFAULT_WORKFLOW_APP_ID = "00000000-0000-0000-0000-000000000002"
@@ -52,6 +51,38 @@ DEFAULT_WORKFLOW_CREATED_BY = "00000000-0000-0000-0000-000000000003"
 type WorkflowVariablePayload = dict[str, object]
 
 
+@dataclass(frozen=True)
+class SQLiteDatabase:
+    """Expose the concrete SQLite engine and scoped session interface used by controller code."""
+
+    engine: Engine
+    session: scoped_session[Session]
+
+
+@pytest.fixture(autouse=True)
+def sqlite_database(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_engine: Engine,
+) -> Iterator[scoped_session[Session]]:
+    """Route controller transactions and model author lookups through SQLite."""
+
+    database_session = scoped_session(sessionmaker(bind=sqlite_engine, expire_on_commit=False))
+    database = SQLiteDatabase(engine=sqlite_engine, session=database_session)
+    monkeypatch.setattr(workflow_controller, "db", database)
+    monkeypatch.setattr(workflow_models, "db", database)
+
+    with database_session() as session:
+        default_author = Account(name="Default Author", email="default-author@example.com")
+        default_author.id = DEFAULT_WORKFLOW_CREATED_BY
+        session.add(default_author)
+        session.commit()
+
+    try:
+        yield database_session
+    finally:
+        database_session.remove()
+
+
 def empty_mapping() -> dict[str, object]:
     return {}
 
@@ -206,18 +237,15 @@ def make_pipeline(
 
 
 @pytest.fixture
-def workflow_author(db_session_with_containers: Session) -> Account:
+def workflow_author(sqlite_database: scoped_session[Session]) -> Account:
     account = Account(name="Alice", email=f"alice-{uuid4()}@example.com")
-    db_session_with_containers.add(account)
-    db_session_with_containers.commit()
+    account.id = str(uuid4())
+    sqlite_database.add(account)
+    sqlite_database.commit()
     return account
 
 
 class TestDraftWorkflowApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_get_draft_success(self, app: Flask, workflow_author: Account) -> None:
         api = DraftRagPipelineApi()
         method = unwrap(api.get)
@@ -278,7 +306,6 @@ class TestDraftWorkflowApi:
 
         with (
             app.test_request_context("/", json={"graph": empty_mapping(), "features": empty_mapping()}),
-            patch.object(type(console_ns), "payload", {"graph": empty_mapping(), "features": empty_mapping()}),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.RagPipelineService",
                 return_value=service,
@@ -373,10 +400,6 @@ class TestDraftWorkflowApi:
 
 
 class TestDraftRunNodes:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_iteration_node_success(self, app: Flask) -> None:
         api = RagPipelineDraftRunIterationNodeApi()
         method = unwrap(api.post)
@@ -386,7 +409,6 @@ class TestDraftRunNodes:
 
         with (
             app.test_request_context("/", json={"inputs": empty_mapping()}),
-            patch.object(type(console_ns), "payload", {"inputs": empty_mapping()}),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate_single_iteration",
                 return_value=MagicMock(),
@@ -408,7 +430,6 @@ class TestDraftRunNodes:
 
         with (
             app.test_request_context("/", json={"inputs": empty_mapping()}),
-            patch.object(type(console_ns), "payload", {"inputs": empty_mapping()}),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate_single_iteration",
                 side_effect=services.errors.conversation.ConversationNotExistsError(),
@@ -426,7 +447,6 @@ class TestDraftRunNodes:
 
         with (
             app.test_request_context("/", json={"inputs": empty_mapping()}),
-            patch.object(type(console_ns), "payload", {"inputs": empty_mapping()}),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate_single_loop",
                 return_value=MagicMock(),
@@ -439,86 +459,7 @@ class TestDraftRunNodes:
             assert method(api, user, pipeline, "node") == {"ok": True}
 
 
-class TestPipelineRunApis:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_draft_run_success(self, app: Flask) -> None:
-        api = DraftRagPipelineRunApi()
-        method = unwrap(api.post)
-
-        pipeline = make_pipeline()
-        user = make_account()
-        session = MagicMock(spec=Session)
-
-        payload = {
-            "inputs": empty_mapping(),
-            "datasource_type": "x",
-            "datasource_info_list": empty_list(),
-            "start_node_id": "n",
-        }
-
-        with (
-            app.test_request_context("/", json=payload),
-            patch.object(type(console_ns), "payload", payload),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.load_rag_pipeline",
-                return_value=pipeline,
-            ) as load_pipeline,
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate",
-                return_value=MagicMock(),
-            ) as generate,
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.helper.compact_generate_response",
-                return_value={"ok": True},
-            ),
-        ):
-            assert method(api, session, user, pipeline.id) == {"ok": True}
-
-        load_pipeline.assert_called_once_with(session, pipeline.id)
-        assert generate.call_args.kwargs["session"] is session
-
-    def test_draft_run_rate_limit(self, app: Flask) -> None:
-        api = DraftRagPipelineRunApi()
-        method = unwrap(api.post)
-
-        pipeline = make_pipeline()
-        user = make_account()
-        session = MagicMock(spec=Session)
-        payload: dict[str, object] = {
-            "inputs": empty_mapping(),
-            "datasource_type": "x",
-            "datasource_info_list": empty_list(),
-            "start_node_id": "n",
-        }
-
-        with (
-            app.test_request_context("/", json=payload),
-            patch.object(
-                type(console_ns),
-                "payload",
-                payload,
-            ),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.load_rag_pipeline",
-                return_value=pipeline,
-            ),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate",
-                side_effect=InvokeRateLimitError("limit"),
-            ),
-        ):
-            with pytest.raises(InvokeRateLimitHttpError):
-                method(api, session, user, pipeline.id)
-
-
 class TestDraftNodeRun:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_execution_not_found(self, app: Flask) -> None:
         api = RagPipelineDraftNodeRunApi()
         method = unwrap(api.post)
@@ -531,7 +472,6 @@ class TestDraftNodeRun:
 
         with (
             app.test_request_context("/", json={"inputs": empty_mapping()}),
-            patch.object(type(console_ns), "payload", {"inputs": empty_mapping()}),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.RagPipelineService",
                 return_value=service,
@@ -542,11 +482,7 @@ class TestDraftNodeRun:
 
 
 class TestPublishedPipelineApis:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_publish_success(self, app: Flask, db_session_with_containers: Session) -> None:
+    def test_publish_success(self, app: Flask) -> None:
         api = PublishedRagPipelineApi()
         method = unwrap(api.post)
 
@@ -557,10 +493,6 @@ class TestPublishedPipelineApis:
             description="test",
             created_by=str(uuid4()),
         )
-        db_session_with_containers.add(pipeline)
-        db_session_with_containers.commit()
-        db_session_with_containers.expire_all()
-
         user = make_account(id="u1")
 
         workflow = make_workflow(id=str(uuid4()), created_at=naive_utc_now())
@@ -582,10 +514,6 @@ class TestPublishedPipelineApis:
 
 
 class TestMiscApis:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_task_stop(self, app: Flask) -> None:
         api = RagPipelineTaskStopApi()
         method = unwrap(api.post)
@@ -603,18 +531,6 @@ class TestMiscApis:
             stop_mock.assert_called_once()
             assert result["result"] == "success"
 
-    def test_transform_forbidden(self, app: Flask) -> None:
-        api = RagPipelineTransformApi()
-        method = unwrap(api.post)
-
-        user = make_account(role=TenantAccountRole.NORMAL)
-
-        with (
-            app.test_request_context("/"),
-        ):
-            with pytest.raises(Forbidden):
-                method(api, MagicMock(spec=Session), user, "ds1")
-
     def test_recommended_plugins(self, app: Flask) -> None:
         api = RagPipelineRecommendedPluginApi()
         method = unwrap(api.get)
@@ -640,85 +556,7 @@ class TestMiscApis:
             service.get_recommended_plugins.assert_called_once_with("all", user, tenant_id)
 
 
-class TestPublishedRagPipelineRunApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_published_run_success(self, app: Flask) -> None:
-        api = PublishedRagPipelineRunApi()
-        method = unwrap(api.post)
-
-        pipeline = make_pipeline()
-        user = make_account()
-        session = MagicMock(spec=Session)
-
-        payload = {
-            "inputs": empty_mapping(),
-            "datasource_type": "x",
-            "datasource_info_list": empty_list(),
-            "start_node_id": "n",
-            "response_mode": "blocking",
-        }
-
-        with (
-            app.test_request_context("/", json=payload),
-            patch.object(type(console_ns), "payload", payload),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.load_rag_pipeline",
-                return_value=pipeline,
-            ) as load_pipeline,
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate",
-                return_value=MagicMock(),
-            ) as generate,
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.helper.compact_generate_response",
-                return_value={"ok": True},
-            ),
-        ):
-            result = method(api, session, user, pipeline.id)
-            assert result == {"ok": True}
-
-        load_pipeline.assert_called_once_with(session, pipeline.id)
-        assert generate.call_args.kwargs["session"] is session
-
-    def test_published_run_rate_limit(self, app: Flask) -> None:
-        api = PublishedRagPipelineRunApi()
-        method = unwrap(api.post)
-
-        pipeline = make_pipeline()
-        user = make_account()
-        session = MagicMock(spec=Session)
-
-        payload = {
-            "inputs": empty_mapping(),
-            "datasource_type": "x",
-            "datasource_info_list": empty_list(),
-            "start_node_id": "n",
-        }
-
-        with (
-            app.test_request_context("/", json=payload),
-            patch.object(type(console_ns), "payload", payload),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.load_rag_pipeline",
-                return_value=pipeline,
-            ),
-            patch(
-                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.PipelineGenerateService.generate",
-                side_effect=InvokeRateLimitError("limit"),
-            ),
-        ):
-            with pytest.raises(InvokeRateLimitHttpError):
-                method(api, session, user, pipeline.id)
-
-
 class TestDefaultBlockConfigApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_get_block_config_success(self, app: Flask) -> None:
         api = DefaultRagPipelineBlockConfigApi()
         method = unwrap(api.get)
@@ -750,10 +588,6 @@ class TestDefaultBlockConfigApi:
 
 
 class TestPublishedAllRagPipelineApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_get_published_workflows_success(self, app: Flask) -> None:
         api = PublishedAllRagPipelineApi()
         method = unwrap(api.get)
@@ -792,10 +626,6 @@ class TestPublishedAllRagPipelineApi:
 
 
 class TestRagPipelineByIdApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_patch_success(self, app: Flask) -> None:
         api = RagPipelineByIdApi()
         method = unwrap(api.patch)
@@ -812,7 +642,6 @@ class TestRagPipelineByIdApi:
 
         with (
             app.test_request_context("/", json=payload),
-            patch.object(type(console_ns), "payload", payload),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.RagPipelineService",
                 return_value=service,
@@ -831,10 +660,7 @@ class TestRagPipelineByIdApi:
         pipeline = make_pipeline()
         user = make_account()
 
-        with (
-            app.test_request_context("/", json={}),
-            patch.object(type(console_ns), "payload", empty_mapping()),
-        ):
+        with app.test_request_context("/", json={}):
             result, status = method(api, user, pipeline, "w1")
             assert status == 400
 
@@ -870,10 +696,6 @@ class TestRagPipelineByIdApi:
 
 
 class TestRagPipelineWorkflowLastRunApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_last_run_success(self, app: Flask) -> None:
         api = RagPipelineWorkflowLastRunApi()
         method = unwrap(api.get)
@@ -919,10 +741,6 @@ class TestRagPipelineWorkflowLastRunApi:
 
 
 class TestRagPipelineWorkflowRunNodeExecutionListApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_get_node_executions_passes_current_user(self, app: Flask) -> None:
         api = RagPipelineWorkflowRunNodeExecutionListApi()
         method = unwrap(api.get)
@@ -955,10 +773,6 @@ class TestRagPipelineWorkflowRunNodeExecutionListApi:
 
 
 class TestRagPipelineDatasourceVariableApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
     def test_set_datasource_variables_success(self, app: Flask) -> None:
         api = RagPipelineDatasourceVariableApi()
         method = unwrap(api.post)
@@ -978,7 +792,6 @@ class TestRagPipelineDatasourceVariableApi:
 
         with (
             app.test_request_context("/", json=payload),
-            patch.object(type(console_ns), "payload", payload),
             patch(
                 "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.RagPipelineService",
                 return_value=service,

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

@@ -273,7 +273,10 @@ class TestChatTextApi:
             ),
             patch.object(audio_module.AudioService, "transcript_tts", transcript_tts),
         ):
-            resp = self.method(installed_app)
+            resp = self.method(
+                audio_module.TextToAudioPayload.model_validate({"message_id": "m1", "text": "hello", "voice": "v1"}),
+                installed_app,
+            )
 
         assert resp == {"audio": "ok"}
         assert transcript_tts.call_args.kwargs["message_ref"] == MessageRef(
@@ -295,7 +298,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(ProviderNotInitializeError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_model_not_supported(self, app: Flask, installed_app):
         with (
@@ -310,7 +313,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(ProviderModelCurrentlyNotSupportError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_invoke_error(self, app: Flask, installed_app):
         with (
@@ -325,7 +328,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(CompletionRequestError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_unknown_exception(self, app: Flask, installed_app):
         with (
@@ -340,7 +343,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(InternalServerError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_app_unavailable_tts(self, app: Flask, installed_app):
         with (
@@ -355,7 +358,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(AppUnavailableError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_no_audio_uploaded_tts(self, app: Flask, installed_app):
         with (
@@ -370,7 +373,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(NoAudioUploadedError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_audio_too_large_tts(self, app: Flask, installed_app):
         with (
@@ -385,7 +388,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(AudioTooLargeError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_unsupported_audio_type_tts(self, app: Flask, installed_app):
         with (
@@ -400,7 +403,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(audio_module.UnsupportedAudioTypeError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_provider_not_support_speech_to_text_tts(self, app: Flask, installed_app):
         with (
@@ -415,7 +418,7 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(audio_module.ProviderNotSupportSpeechToTextError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)
 
     def test_quota_exceeded_tts(self, app: Flask, installed_app):
         with (
@@ -430,4 +433,4 @@ class TestChatTextApi:
             ),
         ):
             with pytest.raises(ProviderQuotaExceededError):
-                self.method(installed_app)
+                self.method(audio_module.TextToAudioPayload.model_validate({"text": "hi"}), installed_app)

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

@@ -57,7 +57,7 @@ def payload_patch(payload_data):
 
 
 class TestCompletionApi:
-    def test_post_success(self, app: Flask, completion_app, user, payload_patch):
+    def test_post_success(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -75,7 +75,13 @@ class TestCompletionApi:
                 return_value=("ok", 200),
             ),
         ):
-            result = method(api, MagicMock(), user, completion_app)
+            result = method(
+                api,
+                completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                MagicMock(),
+                user,
+                completion_app,
+            )
 
         assert result == ("ok", 200)
 
@@ -86,9 +92,15 @@ class TestCompletionApi:
         installed_app = _installed_app(AppMode.CHAT)
 
         with pytest.raises(NotCompletionAppError):
-            method(api, MagicMock(), user, installed_app)
+            method(
+                api,
+                completion_module.CompletionMessageExplorePayload.model_validate({"inputs": {}, "query": "hi"}),
+                MagicMock(),
+                user,
+                installed_app,
+            )
 
-    def test_conversation_completed(self, app: Flask, completion_app, user, payload_patch):
+    def test_conversation_completed(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -102,9 +114,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(ConversationCompletedError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_internal_error(self, app: Flask, completion_app, user, payload_patch):
+    def test_internal_error(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -118,9 +136,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(InternalServerError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_conversation_not_exists(self, app: Flask, completion_app, user, payload_patch):
+    def test_conversation_not_exists(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -134,9 +158,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.NotFound):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_app_unavailable(self, app: Flask, completion_app, user, payload_patch):
+    def test_app_unavailable(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -150,9 +180,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.AppUnavailableError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_provider_not_initialized(self, app: Flask, completion_app, user, payload_patch):
+    def test_provider_not_initialized(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -166,9 +202,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderNotInitializeError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_quota_exceeded(self, app: Flask, completion_app, user, payload_patch):
+    def test_quota_exceeded(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -182,9 +224,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderQuotaExceededError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_model_not_supported(self, app: Flask, completion_app, user, payload_patch):
+    def test_model_not_supported(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -198,9 +246,15 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderModelCurrentlyNotSupportError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
-    def test_invoke_error(self, app: Flask, completion_app, user, payload_patch):
+    def test_invoke_error(self, app: Flask, completion_app, user, payload_patch, payload_data):
         api = completion_module.CompletionApi()
         method = unwrap(api.post)
 
@@ -214,7 +268,13 @@ class TestCompletionApi:
             ),
         ):
             with pytest.raises(completion_module.CompletionRequestError):
-                method(api, MagicMock(), user, completion_app)
+                method(
+                    api,
+                    completion_module.CompletionMessageExplorePayload.model_validate(payload_data),
+                    MagicMock(),
+                    user,
+                    completion_app,
+                )
 
 
 class TestCompletionStopApi:
@@ -239,7 +299,7 @@ class TestCompletionStopApi:
 
 
 class TestChatApi:
-    def test_post_success(self, app: Flask, chat_app, user, payload_patch):
+    def test_post_success(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -257,7 +317,9 @@ class TestChatApi:
                 return_value=("ok", 200),
             ),
         ):
-            result = method(api, MagicMock(), user, chat_app)
+            result = method(
+                api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+            )
 
         assert result == ("ok", 200)
 
@@ -268,9 +330,15 @@ class TestChatApi:
         installed_app = _installed_app(AppMode.COMPLETION)
 
         with pytest.raises(NotChatAppError):
-            method(api, MagicMock(), user, installed_app)
+            method(
+                api,
+                completion_module.ChatMessagePayload.model_validate({"inputs": {}, "query": "hi"}),
+                MagicMock(),
+                user,
+                installed_app,
+            )
 
-    def test_rate_limit_error(self, app: Flask, chat_app, user, payload_patch):
+    def test_rate_limit_error(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -284,9 +352,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(InvokeRateLimitHttpError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_conversation_completed_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_conversation_completed_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -300,9 +370,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(ConversationCompletedError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_conversation_not_exists_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_conversation_not_exists_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -316,7 +388,9 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.NotFound):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
     def test_invalid_conversation_id_fails_fast_as_not_found(self, app: Flask, chat_app, user) -> None:
         # A nonexistent conversation_id must fail fast as 404, before the streaming
@@ -349,13 +423,21 @@ class TestChatApi:
             patch.object(completion_module.AppGenerateService, "generate", generate_mock),
         ):
             with pytest.raises(completion_module.NotFound):
-                method(api, session, user, chat_app)
+                method(
+                    api,
+                    completion_module.ChatMessagePayload.model_validate(
+                        {"inputs": {}, "query": "hi", "conversation_id": conversation_id}
+                    ),
+                    session,
+                    user,
+                    chat_app,
+                )
 
         # The lookup must run before generation, so the generator is never started.
         generate_mock.assert_not_called()
         assert get_conversation_mock.call_args.kwargs["session"] is session
 
-    def test_app_unavailable_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_app_unavailable_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -369,9 +451,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.AppUnavailableError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_provider_not_initialized_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_provider_not_initialized_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -385,9 +469,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderNotInitializeError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_quota_exceeded_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_quota_exceeded_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -401,9 +487,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderQuotaExceededError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_model_not_supported_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_model_not_supported_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -417,9 +505,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.ProviderModelCurrentlyNotSupportError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_invoke_error_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_invoke_error_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -433,9 +523,11 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(completion_module.CompletionRequestError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
-    def test_internal_error_chat(self, app: Flask, chat_app, user, payload_patch):
+    def test_internal_error_chat(self, app: Flask, chat_app, user, payload_patch, payload_data):
         api = completion_module.ChatApi()
         method = unwrap(api.post)
 
@@ -449,7 +541,9 @@ class TestChatApi:
             ),
         ):
             with pytest.raises(InternalServerError):
-                method(api, MagicMock(), user, chat_app)
+                method(
+                    api, completion_module.ChatMessagePayload.model_validate(payload_data), MagicMock(), user, chat_app
+                )
 
 
 class TestChatStopApi:

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

@@ -548,7 +548,7 @@ class TestInstalledAppsCreateApi:
             payload_patch({"app_id": "a1"}),
             patch.object(module.db, "session", session),
         ):
-            result = method(api, tenant_id)
+            result = method(api, module.InstalledAppCreatePayload.model_validate({"app_id": "a1"}), tenant_id)
 
         assert result == {"message": "App installed successfully"}
         assert recommended.install_count == 1
@@ -566,7 +566,7 @@ class TestInstalledAppsCreateApi:
             patch.object(module.db, "session", session),
         ):
             with pytest.raises(NotFound):
-                method(api, tenant_id)
+                method(api, module.InstalledAppCreatePayload.model_validate({"app_id": "a1"}), tenant_id)
 
     def test_post_app_not_public(self, app: Flask, tenant_id: str, payload_patch: PayloadPatch) -> None:
         api = module.InstalledAppsListApi()
@@ -587,7 +587,7 @@ class TestInstalledAppsCreateApi:
             patch.object(module.db, "session", session),
         ):
             with pytest.raises(Forbidden):
-                method(api, tenant_id)
+                method(api, module.InstalledAppCreatePayload.model_validate({"app_id": "a1"}), tenant_id)
 
 
 class TestInstalledAppApi:
@@ -665,7 +665,7 @@ class TestInstalledAppApi:
             payload_patch({"is_pinned": True}),
             patch.object(module.db, "session"),
         ):
-            result = method(installed_app)
+            result = method(api, module.InstalledAppUpdatePayload.model_validate({"is_pinned": True}), installed_app)
 
         assert installed_app.is_pinned is True
         assert result["result"] == "success"
@@ -675,7 +675,7 @@ class TestInstalledAppApi:
         method = unwrap(api.patch)
 
         with app.test_request_context("/", json={}), payload_patch({}), patch.object(module.db, "session"):
-            result = method(installed_app)
+            result = method(api, module.InstalledAppUpdatePayload.model_validate({}), installed_app)
 
         assert result["result"] == "success"
 
@@ -869,7 +869,9 @@ def test_sqlite_post_installs_public_recommended_app_and_is_idempotent(
             payload_patch({"app_id": app_model.id}),
             patch.object(module.db, "session", database),
         ):
-            assert method(api, tenant_id) == {"message": "App installed successfully"}
+            assert method(
+                api, module.InstalledAppCreatePayload.model_validate({"app_id": app_model.id}), tenant_id
+            ) == {"message": "App installed successfully"}
 
     # End the request-scoped session so this assertion only observes committed data.
     bind = database.get_bind()
@@ -899,7 +901,7 @@ def test_sqlite_post_enforces_recommendation_and_public_state(
         patch.object(module.db, "session", database),
         pytest.raises(NotFound),
     ):
-        method(api, tenant_id)
+        method(api, module.InstalledAppCreatePayload.model_validate({"app_id": "missing"}), tenant_id)
 
     private_app = _persist_app(database, app_id="private", public=False)
     database.add(
@@ -918,7 +920,7 @@ def test_sqlite_post_enforces_recommendation_and_public_state(
         patch.object(module.db, "session", database),
         pytest.raises(Forbidden),
     ):
-        method(api, tenant_id)
+        method(api, module.InstalledAppCreatePayload.model_validate({"app_id": private_app.id}), tenant_id)
 
 
 def test_sqlite_delete_removes_foreign_installed_app_and_rejects_owned_app(
@@ -957,7 +959,12 @@ def test_sqlite_patch_persists_pin_and_noop_payload(
         payload_patch({"is_pinned": True}),
         patch.object(module.db, "session", database),
     ):
-        assert unwrap(api.patch)(installed)["result"] == "success"
+        assert (
+            unwrap(api.patch)(api, module.InstalledAppUpdatePayload.model_validate({"is_pinned": True}), installed)[
+                "result"
+            ]
+            == "success"
+        )
     database.expire_all()
     persisted_installed_app = database.get(InstalledApp, installed.id)
     assert persisted_installed_app is not None
@@ -968,4 +975,7 @@ def test_sqlite_patch_persists_pin_and_noop_payload(
         payload_patch({}),
         patch.object(module.db, "session", database),
     ):
-        assert unwrap(api.patch)(installed)["result"] == "success"
+        assert (
+            unwrap(api.patch)(api, module.InstalledAppUpdatePayload.model_validate({}), installed)["result"]
+            == "success"
+        )

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

@@ -160,7 +160,9 @@ class TestMessageFeedbackApi:
                 "create_feedback",
             ),
         ):
-            result = method(MagicMock(), installed_app, "mid")
+            result = method(
+                module.MessageFeedbackPayload.model_validate({"rating": "like"}), MagicMock(), installed_app, "mid"
+            )
 
         assert result["result"] == "success"
 
@@ -179,7 +181,7 @@ class TestMessageFeedbackApi:
             ),
         ):
             with pytest.raises(NotFound):
-                method(MagicMock(), installed_app, "mid")
+                method(module.MessageFeedbackPayload.model_validate({}), MagicMock(), installed_app, "mid")
 
 
 class TestMessageMoreLikeThisApi:

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

@@ -100,7 +100,7 @@ class TestSavedMessageListApi:
             payload_patch(payload),
             patch.object(module.SavedMessageService, "save") as save_mock,
         ):
-            result = method(api, current_user, installed_app)
+            result = method(api, module.SavedMessageCreatePayload.model_validate(payload), current_user, installed_app)
 
         save_mock.assert_called_once()
         assert save_mock.call_args.args[1] is current_user
@@ -124,7 +124,7 @@ class TestSavedMessageListApi:
             ),
         ):
             with pytest.raises(NotFound):
-                method(api, MagicMock(), installed_app)
+                method(api, module.SavedMessageCreatePayload.model_validate(payload), MagicMock(), installed_app)
 
 
 class TestSavedMessageApi:

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

@@ -5,6 +5,7 @@ import pytest
 from flask import Flask
 from werkzeug.exceptions import InternalServerError
 
+from controllers.common.controller_schemas import WorkflowRunPayload
 from controllers.console.explore.error import NotWorkflowAppError
 from controllers.console.explore.workflow import (
     InstalledAppWorkflowRunApi,
@@ -62,11 +63,18 @@ class TestInstalledAppWorkflowRunApi:
 
         with app.test_request_context("/"):
             with pytest.raises(NotWorkflowAppError):
-                method(api, MagicMock(), MagicMock(), non_workflow_installed_app)
+                method(
+                    api,
+                    WorkflowRunPayload.model_validate({"inputs": {}}),
+                    MagicMock(),
+                    MagicMock(),
+                    non_workflow_installed_app,
+                )
 
     def test_success(self, app: Flask, installed_workflow_app, user, payload):
         api = InstalledAppWorkflowRunApi()
         method = unwrap(api.post)
+        req_data = WorkflowRunPayload.model_validate(payload)
 
         with (
             app.test_request_context("/", json=payload),
@@ -75,7 +83,7 @@ class TestInstalledAppWorkflowRunApi:
                 return_value=MagicMock(),
             ) as generate_mock,
         ):
-            result = method(api, MagicMock(), user, installed_workflow_app)
+            result = method(api, req_data, MagicMock(), user, installed_workflow_app)
 
             generate_mock.assert_called_once()
             assert generate_mock.call_args.kwargs["user"] is user
@@ -84,6 +92,7 @@ class TestInstalledAppWorkflowRunApi:
     def test_rate_limit_error(self, app: Flask, installed_workflow_app, user, payload):
         api = InstalledAppWorkflowRunApi()
         method = unwrap(api.post)
+        req_data = WorkflowRunPayload.model_validate(payload)
 
         with (
             app.test_request_context("/", json=payload),
@@ -93,11 +102,12 @@ class TestInstalledAppWorkflowRunApi:
             ),
         ):
             with pytest.raises(InvokeRateLimitHttpError):
-                method(api, MagicMock(), user, installed_workflow_app)
+                method(api, req_data, MagicMock(), user, installed_workflow_app)
 
     def test_unexpected_exception(self, app: Flask, installed_workflow_app, user, payload):
         api = InstalledAppWorkflowRunApi()
         method = unwrap(api.post)
+        req_data = WorkflowRunPayload.model_validate(payload)
 
         with (
             app.test_request_context("/", json=payload),
@@ -107,7 +117,7 @@ class TestInstalledAppWorkflowRunApi:
             ),
         ):
             with pytest.raises(InternalServerError):
-                method(api, MagicMock(), user, installed_workflow_app)
+                method(api, req_data, MagicMock(), user, installed_workflow_app)
 
 
 class TestInstalledAppWorkflowTaskStopApi:

api/tests/unit_tests/controllers/console/test_fastopenapi_setup.py

@@ -1,40 +1,84 @@
 import builtins
-from unittest.mock import patch
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import Mock, create_autospec
 
 import pytest
-from flask import Flask
 from flask.views import MethodView
 
+from controllers.console import setup as setup_controller
+from controllers.console import wraps
+from controllers.console.error import AlreadySetupError, NotInitValidateError
+from dify_app import DifyApp
 from extensions import ext_fastopenapi
+from services.setup_service import (
+    InitializationValidationRequiredError,
+    SetupAlreadyCompletedError,
+    SetupInput,
+    SetupService,
+    SetupStatus,
+)
 
 if not hasattr(builtins, "MethodView"):
     builtins.MethodView = MethodView  # type: ignore[attr-defined]
 
 
 @pytest.fixture
-def app() -> Flask:
-    app = Flask(__name__)
+def setup_service(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    service = create_autospec(SetupService, instance=True, spec_set=True)
+    services = SimpleNamespace(setup=service)
+    monkeypatch.setattr(setup_controller, "application_services", lambda: services)
+    return service
+
+
+@pytest.fixture
+def app() -> DifyApp:
+    app = DifyApp(__name__)
     app.config["TESTING"] = True
+    ext_fastopenapi.init_app(app)
     return app
 
 
-def test_console_setup_fastopenapi_get_not_started(app: Flask):
-    ext_fastopenapi.init_app(app)
+def test_console_setup_fastopenapi_get_not_started(app: DifyApp, setup_service: Mock) -> None:
+    setup_service.get_status.return_value = SetupStatus(completed=False)
 
-    with (
-        patch("controllers.console.setup.dify_config.EDITION", "SELF_HOSTED"),
-        patch("controllers.console.setup.get_setup_status", return_value=None),
-    ):
-        client = app.test_client()
-        response = client.get("/console/api/setup")
+    response = app.test_client().get("/console/api/setup")
 
     assert response.status_code == 200
     assert response.get_json() == {"step": "not_started", "setup_at": None}
 
 
-def test_console_setup_fastopenapi_post_success(app: Flask):
-    ext_fastopenapi.init_app(app)
+def test_console_setup_fastopenapi_get_finished(app: DifyApp, setup_service: Mock) -> None:
+    setup_at = datetime(2026, 8, 6, 10, 30)
+    setup_service.get_status.return_value = SetupStatus(completed=True, setup_at=setup_at)
 
+    response = app.test_client().get("/console/api/setup")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"step": "finished", "setup_at": "2026-08-06T10:30:00"}
+
+
+def test_console_setup_fastopenapi_get_finished_without_setup_time(app: DifyApp, setup_service: Mock) -> None:
+    setup_service.get_status.return_value = SetupStatus(completed=True)
+
+    response = app.test_client().get("/console/api/setup")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"step": "finished", "setup_at": None}
+
+
+@pytest.mark.parametrize("enterprise_enabled", [False, True], ids=["community", "enterprise"])
+def test_console_setup_fastopenapi_post_success(
+    app: DifyApp,
+    setup_service: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    enterprise_enabled: bool,
+) -> None:
+    monkeypatch.setattr(wraps.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(wraps.dify_config, "ENTERPRISE_ENABLED", enterprise_enabled)
+    monkeypatch.setattr(setup_controller, "get_init_validate_status", lambda: True)
+    mark_setup_completed = Mock()
+    monkeypatch.setattr(setup_controller, "mark_setup_completed", mark_setup_completed)
     payload = {
         "email": "admin@example.com",
         "name": "Admin",
@@ -42,17 +86,162 @@ def test_console_setup_fastopenapi_post_success(app: Flask):
         "language": "en-US",
     }
 
-    with (
-        patch("controllers.console.wraps.dify_config.EDITION", "SELF_HOSTED"),
-        patch("controllers.console.setup.get_setup_status", return_value=None),
-        patch("controllers.console.setup.TenantService.get_tenant_count", return_value=0),
-        patch("controllers.console.setup.get_init_validate_status", return_value=True),
-        patch("controllers.console.setup.RegisterService.setup"),
-        patch("controllers.console.setup.mark_setup_completed") as mark_setup_completed,
-    ):
-        client = app.test_client()
-        response = client.post("/console/api/setup", json=payload)
+    response = app.test_client().post(
+        "/console/api/setup",
+        json=payload,
+        headers={"CF-Connecting-IP": "203.0.113.7"},
+    )
 
     assert response.status_code == 201
     assert response.get_json() == {"result": "success"}
+    setup_service.initialize.assert_called_once_with(
+        SetupInput(
+            email="admin@example.com",
+            name="Admin",
+            password="Passw0rd1",
+            ip_address="203.0.113.7",
+            language="en-US",
+        ),
+        initialization_validated=True,
+    )
     mark_setup_completed.assert_called_once_with()
+
+
+def test_console_setup_fastopenapi_post_rejects_cloud_edition(
+    app: DifyApp,
+    setup_service: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(wraps.dify_config, "EDITION", "CLOUD")
+    monkeypatch.setattr(wraps.dify_config, "ENTERPRISE_ENABLED", False)
+
+    response = app.test_client().post(
+        "/console/api/setup",
+        json={
+            "email": "admin@example.com",
+            "name": "Admin",
+            "password": "Passw0rd1",
+            "language": "en-US",
+        },
+    )
+
+    assert response.status_code == 404
+    setup_service.initialize.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        pytest.param(
+            {
+                "email": "not-an-email",
+                "name": "Admin",
+                "password": "Passw0rd1",
+                "language": "en-US",
+            },
+            id="invalid-email",
+        ),
+        pytest.param(
+            {
+                "email": "admin@example.com",
+                "name": "Admin",
+                "password": "short",
+                "language": "en-US",
+            },
+            id="invalid-password",
+        ),
+        pytest.param(
+            {
+                "email": "admin@example.com",
+                "name": "a" * 31,
+                "password": "Passw0rd1",
+                "language": "en-US",
+            },
+            id="name-too-long",
+        ),
+    ],
+)
+def test_console_setup_fastopenapi_post_rejects_invalid_payload_before_service_call(
+    app: DifyApp,
+    setup_service: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    payload: dict[str, str],
+) -> None:
+    monkeypatch.setattr(wraps.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(wraps.dify_config, "ENTERPRISE_ENABLED", False)
+
+    response = app.test_client().post("/console/api/setup", json=payload)
+
+    assert response.status_code == 422
+    setup_service.initialize.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("service_error", "expected_controller_error"),
+    [
+        pytest.param(SetupAlreadyCompletedError(), AlreadySetupError, id="already-setup"),
+        pytest.param(
+            InitializationValidationRequiredError(),
+            NotInitValidateError,
+            id="init-validation-required",
+        ),
+    ],
+)
+def test_console_setup_translates_service_errors_to_controller_errors(
+    app: DifyApp,
+    setup_service: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    service_error: Exception,
+    expected_controller_error: type[Exception],
+) -> None:
+    monkeypatch.setattr(wraps.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(wraps.dify_config, "ENTERPRISE_ENABLED", False)
+    monkeypatch.setattr(setup_controller, "get_init_validate_status", lambda: False)
+    mark_setup_completed = Mock()
+    monkeypatch.setattr(setup_controller, "mark_setup_completed", mark_setup_completed)
+    setup_service.initialize.side_effect = service_error
+
+    payload = setup_controller.SetupRequestPayload.model_validate(
+        {
+            "email": "admin@example.com",
+            "name": "Admin",
+            "password": "Passw0rd1",
+            "language": "en-US",
+        }
+    )
+    with app.test_request_context(
+        "/console/api/setup",
+        method="POST",
+        headers={"CF-Connecting-IP": "203.0.113.7"},
+    ):
+        with pytest.raises(expected_controller_error) as raised:
+            setup_controller.setup_system(payload)
+
+    assert type(raised.value) is expected_controller_error
+    mark_setup_completed.assert_not_called()
+
+
+def test_console_setup_fastopenapi_does_not_mark_setup_completed_when_service_fails(
+    app: DifyApp,
+    setup_service: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(wraps.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(wraps.dify_config, "ENTERPRISE_ENABLED", False)
+    monkeypatch.setattr(setup_controller, "get_init_validate_status", lambda: True)
+    mark_setup_completed = Mock()
+    monkeypatch.setattr(setup_controller, "mark_setup_completed", mark_setup_completed)
+    setup_service.initialize.side_effect = RuntimeError("provision failed")
+
+    response = app.test_client().post(
+        "/console/api/setup",
+        json={
+            "email": "admin@example.com",
+            "name": "Admin",
+            "password": "Passw0rd1",
+            "language": "en-US",
+        },
+    )
+
+    assert response.status_code == 500
+    mark_setup_completed.assert_not_called()

api/tests/unit_tests/controllers/console/test_feature.py

@@ -16,6 +16,8 @@ from services.entities.feature_entities import (
     VectorSpaceLimitationModel,
 )
 from services.feature_query_service import FeatureQueryService
+from services.schema_definition_service import SchemaDefinitionService
+from services.setup_service import SetupService
 from services.workspace_member_query_service import WorkspaceMemberQueryService
 from services.workspace_query_service import WorkspaceQueryService
 
@@ -32,6 +34,8 @@ def _request_context() -> RequestContext:
 def _install_application_services(mocker: MockerFixture):
     feature_queries = create_autospec(FeatureQueryService, instance=True, spec_set=True)
     services = ApplicationServices(
+        schema_definitions=create_autospec(SchemaDefinitionService, instance=True, spec_set=True),
+        setup=create_autospec(SetupService, instance=True, spec_set=True),
         feature_queries=feature_queries,
         workspace_queries=create_autospec(WorkspaceQueryService, instance=True, spec_set=True),
         workspace_member_queries=create_autospec(WorkspaceMemberQueryService, instance=True, spec_set=True),

api/tests/unit_tests/controllers/console/test_spec.py

@@ -1,15 +1,24 @@
 from inspect import unwrap
-from unittest.mock import patch
-
-import pytest
+from types import SimpleNamespace
+from unittest.mock import create_autospec, patch
 
 import controllers.console.spec as spec_module
+from dify_app import DifyApp
+from extensions import ext_login
+from machinery.context import RequestContext
+from services.schema_definition_service import SchemaDefinitionService
 
 
 class TestSpecSchemaDefinitionsApi:
-    def test_get_success(self):
+    def test_get_success(self) -> None:
         api = spec_module.SpecSchemaDefinitionsApi()
         method = unwrap(api.get)
+        request_context = RequestContext(
+            request_id="request-1",
+            trace_id="trace-1",
+            account_id="account-1",
+            active_workspace_id="workspace-1",
+        )
 
         schema_definitions = [
             {
@@ -23,34 +32,65 @@ class TestSpecSchemaDefinitionsApi:
             }
         ]
 
+        service = create_autospec(SchemaDefinitionService, instance=True, spec_set=True)
+        service.list.return_value = tuple(schema_definitions)
+
         with patch.object(
             spec_module,
-            "SchemaManager",
-        ) as schema_manager_cls:
-            schema_manager_cls.return_value.get_all_schema_definitions.return_value = schema_definitions
+            "application_services",
+            return_value=SimpleNamespace(schema_definitions=service),
+        ):
+            resp, status = method(api, request_context)
 
-            resp, status = method(api)
-
-        assert status == 200
+        assert status == spec_module.HTTPStatus.OK
         assert resp == schema_definitions
-        assert spec_module.SchemaDefinitionsResponse.model_validate(resp).model_dump(mode="json") == schema_definitions
+        service.list.assert_called_once_with()
 
-    def test_get_documents_tight_response_model(self):
+    def test_get_documents_tight_response_model(self) -> None:
         response = spec_module.SpecSchemaDefinitionsApi.get.__apidoc__["responses"]["200"]
 
         assert response[1].name == spec_module.SchemaDefinitionsResponse.__name__
 
-    def test_get_exception_returns_empty_list(self, caplog: pytest.LogCaptureFixture):
+    def test_get_returns_empty_list_from_service(self) -> None:
         api = spec_module.SpecSchemaDefinitionsApi()
         method = unwrap(api.get)
+        request_context = RequestContext(
+            request_id="request-1",
+            trace_id=None,
+            account_id="account-1",
+            active_workspace_id=None,
+        )
+        service = create_autospec(SchemaDefinitionService, instance=True, spec_set=True)
+        service.list.return_value = ()
 
         with patch.object(
             spec_module,
-            "SchemaManager",
-            side_effect=Exception("boom"),
+            "application_services",
+            return_value=SimpleNamespace(schema_definitions=service),
         ):
-            resp, status = method(api)
+            resp, status = method(api, request_context)
 
-        assert status == 200
+        assert status == spec_module.HTTPStatus.OK
         assert resp == []
-        assert "boom" in caplog.text
+
+    def test_get_rejects_unauthenticated_request_before_service_call(self) -> None:
+        app = DifyApp(__name__)
+        app.config["TESTING"] = True
+        ext_login.init_app(app)
+        api = spec_module.SpecSchemaDefinitionsApi()
+        service = create_autospec(SchemaDefinitionService, instance=True, spec_set=True)
+
+        with (
+            app.test_request_context("/console/api/spec/schema-definitions"),
+            patch("controllers.console.wraps._is_setup_completed", return_value=True),
+            patch("libs.login._resolve_current_user", return_value=None),
+            patch.object(
+                spec_module,
+                "application_services",
+                return_value=SimpleNamespace(schema_definitions=service),
+            ),
+        ):
+            response = api.get()
+
+        assert response.status_code == 401
+        service.list.assert_not_called()

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_runtime_request_builder.py

@@ -744,7 +744,13 @@ def test_build_maps_agent_soul_knowledge_to_knowledge_layer_config():
                                 "conditions": {
                                     "logical_operator": "and",
                                     "conditions": [
-                                        {"name": "category", "comparison_operator": "contains", "value": "auth"}
+                                        {
+                                            "id": "cond-1",
+                                            "metadata_id": "meta-1",
+                                            "name": "category",
+                                            "comparison_operator": "contains",
+                                            "value": "auth",
+                                        }
                                     ],
                                 },
                             },

api/tests/unit_tests/extensions/test_ext_application_services.py

@@ -0,0 +1,58 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy.orm import Session, sessionmaker
+
+from enums.deployment_edition import DeploymentEdition
+from extensions.ext_application_services import build_application_services
+from extensions.ext_redis import RedisClientWrapper
+
+
+@pytest.mark.parametrize(
+    ("deployment_edition", "setup_completed"),
+    [
+        pytest.param(DeploymentEdition.CLOUD, True, id="cloud"),
+        pytest.param(DeploymentEdition.COMMUNITY, False, id="community"),
+        pytest.param(DeploymentEdition.ENTERPRISE, False, id="enterprise"),
+    ],
+)
+def test_build_application_services_configures_setup_policy(
+    sqlite_session_factory: sessionmaker[Session],
+    deployment_edition: DeploymentEdition,
+    setup_completed: bool,
+) -> None:
+    services = build_application_services(
+        database_client=sqlite_session_factory,
+        deployment_edition=deployment_edition,
+        redis=MagicMock(spec=RedisClientWrapper),
+    )
+
+    assert services.setup.get_status().completed is setup_completed
+
+
+def test_build_application_services_wires_builtin_schema_definitions(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    services = build_application_services(
+        database_client=sqlite_session_factory,
+        deployment_edition=DeploymentEdition.COMMUNITY,
+        redis=MagicMock(spec=RedisClientWrapper),
+    )
+
+    definitions = services.schema_definitions.list()
+
+    assert definitions
+    assert all({"name", "label", "schema"} <= definition.keys() for definition in definitions)
+
+
+def test_build_application_services_does_not_construct_schema_manager(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    with patch("extensions.ext_application_services.SchemaManager") as schema_manager:
+        build_application_services(
+            database_client=sqlite_session_factory,
+            deployment_edition=DeploymentEdition.COMMUNITY,
+            redis=MagicMock(spec=RedisClientWrapper),
+        )
+
+    schema_manager.assert_not_called()

api/tests/unit_tests/repositories/test_installation_state_repository.py

@@ -0,0 +1,40 @@
+"""Persistence tests for installation setup and tenant-existence state."""
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from models.account import Tenant
+from models.model import DifySetup
+from repositories.installation_state_repository import InstallationStateRepository
+
+
+def test_empty_database_has_no_installation_state(sqlite_session_factory: sessionmaker[Session]) -> None:
+    repository = InstallationStateRepository(client=sqlite_session_factory)
+
+    assert repository.get_setup_at() is None
+    assert repository.is_setup() is False
+    assert repository.has_tenants() is False
+
+
+def test_get_setup_at_returns_persisted_timestamp(
+    sqlite_session: Session,
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    setup = DifySetup(version="test-version")
+    sqlite_session.add(setup)
+    sqlite_session.commit()
+    sqlite_session.refresh(setup)
+    repository = InstallationStateRepository(client=sqlite_session_factory)
+
+    assert repository.get_setup_at() == setup.setup_at
+    assert repository.is_setup() is True
+
+
+def test_has_tenants_detects_existing_tenant(
+    sqlite_session: Session,
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    sqlite_session.add(Tenant(name="Existing workspace"))
+    sqlite_session.commit()
+    repository = InstallationStateRepository(client=sqlite_session_factory)
+
+    assert repository.has_tenants() is True

api/tests/unit_tests/services/agent/test_agent_composer_entities.py

@@ -365,6 +365,51 @@ def test_knowledge_runtime_requirements_block_publish_but_not_draft_save(knowled
         ComposerConfigValidator.validate_publish_payload(publish_payload)
 
 
+def test_manual_metadata_filtering_condition_accepts_composer_ui_identifiers():
+    """The composer's condition editor always sends ``id`` (list row key) and
+    ``metadata_id`` (selected metadata field reference) alongside every
+    condition. Rejecting them as unknown fields broke every save of a manual
+    metadata filter (see GH issue #40169)."""
+    payload = ComposerSavePayload.model_validate(
+        {
+            "variant": ComposerVariant.AGENT_APP,
+            "save_strategy": ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION,
+            "agent_soul": {
+                "knowledge": {
+                    "sets": [
+                        {
+                            "id": "support",
+                            "name": "Support KB",
+                            "datasets": [{"id": "dataset-1"}],
+                            "query": {"mode": "generated_query"},
+                            "retrieval": {"mode": "multiple", "top_k": 4},
+                            "metadata_filtering": {
+                                "mode": "manual",
+                                "conditions": {
+                                    "logical_operator": "and",
+                                    "conditions": [
+                                        {
+                                            "id": "b149eceb-191a-40a2-9f11-61cf21ebd147",
+                                            "metadata_id": "ad6cf326-eadf-46e8-a2d5-9cb892d2cc84",
+                                            "name": "category",
+                                            "comparison_operator": "is",
+                                            "value": "auth",
+                                        }
+                                    ],
+                                },
+                            },
+                        },
+                    ]
+                }
+            },
+        }
+    )
+
+    condition = payload.agent_soul.knowledge.sets[0].metadata_filtering.conditions.conditions[0]
+    assert condition.id == "b149eceb-191a-40a2-9f11-61cf21ebd147"
+    assert condition.metadata_id == "ad6cf326-eadf-46e8-a2d5-9cb892d2cc84"
+
+
 def test_agent_soul_model_config_is_first_class_without_credentials():
     config = AgentSoulConfig(
         model=AgentSoulModelConfig(

api/tests/unit_tests/services/test_schema_definition_service.py

@@ -0,0 +1,57 @@
+from collections.abc import Mapping
+from unittest.mock import Mock, create_autospec
+
+import pytest
+
+from services.schema_definition_service import (
+    SchemaDefinitionService,
+    SchemaDefinitionSource,
+)
+
+
+@pytest.fixture
+def source() -> Mock:
+    return create_autospec(SchemaDefinitionSource, instance=True, spec_set=True)
+
+
+@pytest.fixture
+def source_factory(source: Mock) -> Mock:
+    return Mock(return_value=source)
+
+
+def test_list_returns_schema_definitions(source: Mock, source_factory: Mock) -> None:
+    definitions: list[Mapping[str, object]] = [
+        {
+            "name": "conversation-variable",
+            "label": "Conversation variable",
+            "schema": {"type": "object"},
+        }
+    ]
+    source.get_all_schema_definitions.return_value = definitions
+    service = SchemaDefinitionService(source_factory=source_factory)
+
+    assert service.list() == tuple(definitions)
+    source_factory.assert_called_once_with()
+    source.get_all_schema_definitions.assert_called_once_with()
+
+
+def test_list_returns_empty_tuple_when_source_query_fails(
+    source: Mock,
+    source_factory: Mock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    source.get_all_schema_definitions.side_effect = RuntimeError("boom")
+    service = SchemaDefinitionService(source_factory=source_factory)
+
+    assert service.list() == ()
+    assert "Failed to get schema definitions from local registry" in caplog.text
+    assert "boom" in caplog.text
+
+
+def test_list_returns_empty_tuple_when_source_construction_fails(caplog: pytest.LogCaptureFixture) -> None:
+    source_factory = Mock(side_effect=RuntimeError("construction failed"))
+    service = SchemaDefinitionService(source_factory=source_factory)
+
+    assert service.list() == ()
+    assert "Failed to get schema definitions from local registry" in caplog.text
+    assert "construction failed" in caplog.text

api/tests/unit_tests/services/test_setup_adapters.py

@@ -0,0 +1,74 @@
+from contextlib import nullcontext
+from unittest.mock import ANY, MagicMock, patch
+
+import pytest
+from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import LockError
+from sqlalchemy.orm import Session, sessionmaker
+
+from extensions.ext_redis import RedisClientWrapper
+from services.setup_adapters import RedisSetupLock, RegisterServiceAccountProvisioner
+from services.setup_service import SetupInput
+
+
+def test_provision_delegates_to_register_service_with_managed_session(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    provisioner = RegisterServiceAccountProvisioner(client=sqlite_session_factory)
+    setup = SetupInput(
+        email="admin@example.com",
+        name="Admin",
+        password="Passw0rd1",
+        ip_address="203.0.113.7",
+        language="en-US",
+    )
+
+    with patch("services.setup_adapters.RegisterService.setup") as register:
+        provisioner.provision(setup)
+
+    register.assert_called_once_with(
+        email="admin@example.com",
+        name="Admin",
+        password="Passw0rd1",
+        ip_address="203.0.113.7",
+        language="en-US",
+        session=ANY,
+    )
+    assert isinstance(register.call_args.kwargs["session"], Session)
+
+
+def test_acquire_uses_bounded_distributed_lock() -> None:
+    redis = MagicMock(spec=RedisClientWrapper)
+    redis.lock.return_value = nullcontext()
+    lock = RedisSetupLock(client=redis)
+
+    with lock.acquire():
+        pass
+
+    redis.lock.assert_called_once_with(
+        "setup:initialize",
+        timeout=300,
+        blocking_timeout=300,
+    )
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        pytest.param(LockError("lock acquisition timed out"), id="timeout"),
+        pytest.param(RedisConnectionError("redis unavailable"), id="connection"),
+    ],
+)
+def test_acquire_propagates_distributed_lock_failure(error: Exception) -> None:
+    redis = MagicMock(spec=RedisClientWrapper)
+    lock_context = MagicMock()
+    lock_context.__enter__.side_effect = error
+    redis.lock.return_value = lock_context
+    lock = RedisSetupLock(client=redis)
+
+    with pytest.raises(type(error), match=str(error)) as raised:
+        with lock.acquire():
+            pytest.fail("lock body must not run")
+
+    assert raised.value is error
+    lock_context.__exit__.assert_not_called()

api/tests/unit_tests/services/test_setup_service.py

@@ -0,0 +1,249 @@
+from collections.abc import Generator
+from contextlib import contextmanager, nullcontext
+from dataclasses import dataclass
+from datetime import datetime
+from unittest.mock import Mock, create_autospec
+
+import pytest
+
+from services.setup_service import (
+    InitializationValidationRequiredError,
+    SetupAccountProvisioner,
+    SetupAlreadyCompletedError,
+    SetupInput,
+    SetupLock,
+    SetupService,
+    SetupState,
+    SetupStatus,
+)
+
+
+@pytest.fixture
+def state() -> Mock:
+    state = create_autospec(SetupState, instance=True, spec_set=True)
+    state.get_setup_at.return_value = None
+    state.has_tenants.return_value = False
+    return state
+
+
+@pytest.fixture
+def accounts() -> Mock:
+    return create_autospec(SetupAccountProvisioner, instance=True, spec_set=True)
+
+
+@pytest.fixture
+def lock() -> Mock:
+    lock = create_autospec(SetupLock, instance=True, spec_set=True)
+    lock.acquire.return_value = nullcontext()
+    return lock
+
+
+@pytest.fixture
+def setup_input() -> SetupInput:
+    return SetupInput(
+        email="Admin@Example.com",
+        name="Admin",
+        password="Passw0rd1",
+        ip_address="203.0.113.7",
+        language="en-US",
+    )
+
+
+@dataclass
+class TrackingLock:
+    inside: bool = False
+    exited: bool = False
+
+    @contextmanager
+    def acquire(self) -> Generator[None]:
+        self.inside = True
+        try:
+            yield
+        finally:
+            self.inside = False
+            self.exited = True
+
+
+class FailingLock:
+    def __init__(self, error: Exception) -> None:
+        self._error = error
+
+    @contextmanager
+    def acquire(self) -> Generator[None]:
+        raise self._error
+        yield
+
+
+def test_cloud_status_is_finished_without_reading_persistence(state: Mock, accounts: Mock, lock: Mock) -> None:
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=False)
+
+    assert service.get_status() == SetupStatus(completed=True)
+    state.get_setup_at.assert_not_called()
+
+
+def test_self_hosted_status_is_not_started_without_setup(state: Mock, accounts: Mock, lock: Mock) -> None:
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    assert service.get_status() == SetupStatus(completed=False)
+
+
+def test_self_hosted_status_includes_setup_time(state: Mock, accounts: Mock, lock: Mock) -> None:
+    setup_at = datetime(2026, 8, 6, 10, 30)
+    state.get_setup_at.return_value = setup_at
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    assert service.get_status() == SetupStatus(completed=True, setup_at=setup_at)
+
+
+def test_initialize_rejects_existing_setup(
+    state: Mock,
+    accounts: Mock,
+    lock: Mock,
+    setup_input: SetupInput,
+) -> None:
+    state.get_setup_at.return_value = datetime(2026, 8, 6, 10, 30)
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    with pytest.raises(SetupAlreadyCompletedError):
+        service.initialize(setup_input, initialization_validated=True)
+
+    state.has_tenants.assert_not_called()
+    accounts.provision.assert_not_called()
+
+
+def test_initialize_rejects_existing_tenant(
+    state: Mock,
+    accounts: Mock,
+    lock: Mock,
+    setup_input: SetupInput,
+) -> None:
+    state.has_tenants.return_value = True
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    with pytest.raises(SetupAlreadyCompletedError):
+        service.initialize(setup_input, initialization_validated=True)
+
+    accounts.provision.assert_not_called()
+
+
+@pytest.mark.parametrize("persistent_state", ["setup", "tenant"])
+def test_initialize_prioritizes_existing_persistent_state_over_validation(
+    state: Mock,
+    accounts: Mock,
+    lock: Mock,
+    setup_input: SetupInput,
+    persistent_state: str,
+) -> None:
+    if persistent_state == "setup":
+        state.get_setup_at.return_value = datetime(2026, 8, 6, 10, 30)
+    else:
+        state.has_tenants.return_value = True
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    with pytest.raises(SetupAlreadyCompletedError):
+        service.initialize(setup_input, initialization_validated=False)
+
+    accounts.provision.assert_not_called()
+
+
+def test_initialize_requires_initialization_validation(
+    state: Mock,
+    accounts: Mock,
+    lock: Mock,
+    setup_input: SetupInput,
+) -> None:
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    with pytest.raises(InitializationValidationRequiredError):
+        service.initialize(setup_input, initialization_validated=False)
+
+    accounts.provision.assert_not_called()
+
+
+def test_initialize_normalizes_email_and_provisions_account(
+    state: Mock,
+    accounts: Mock,
+    lock: Mock,
+    setup_input: SetupInput,
+) -> None:
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    service.initialize(setup_input, initialization_validated=True)
+
+    accounts.provision.assert_called_once_with(
+        SetupInput(
+            email="admin@example.com",
+            name="Admin",
+            password="Passw0rd1",
+            ip_address="203.0.113.7",
+            language="en-US",
+        )
+    )
+    lock.acquire.assert_called_once_with()
+
+
+def test_initialize_reads_and_writes_only_while_holding_lock(
+    state: Mock,
+    accounts: Mock,
+    setup_input: SetupInput,
+) -> None:
+    lock = TrackingLock()
+
+    def get_setup_at() -> None:
+        assert lock.inside
+
+    def has_tenants() -> bool:
+        assert lock.inside
+        return False
+
+    def provision(_setup: SetupInput) -> None:
+        assert lock.inside
+
+    state.get_setup_at.side_effect = get_setup_at
+    state.has_tenants.side_effect = has_tenants
+    accounts.provision.side_effect = provision
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    service.initialize(setup_input, initialization_validated=True)
+
+    assert lock.exited
+
+
+def test_initialize_does_not_read_or_write_when_lock_acquisition_fails(
+    state: Mock,
+    accounts: Mock,
+    setup_input: SetupInput,
+) -> None:
+    error = TimeoutError("lock acquisition timed out")
+    service = SetupService(
+        state=state,
+        accounts=accounts,
+        lock=FailingLock(error),
+        setup_required=True,
+    )
+
+    with pytest.raises(TimeoutError, match="lock acquisition timed out") as raised:
+        service.initialize(setup_input, initialization_validated=True)
+
+    assert raised.value is error
+    state.get_setup_at.assert_not_called()
+    state.has_tenants.assert_not_called()
+    accounts.provision.assert_not_called()
+
+
+def test_initialize_releases_lock_and_propagates_provision_failure(
+    state: Mock,
+    accounts: Mock,
+    setup_input: SetupInput,
+) -> None:
+    lock = TrackingLock()
+    error = RuntimeError("provision failed")
+    accounts.provision.side_effect = error
+    service = SetupService(state=state, accounts=accounts, lock=lock, setup_required=True)
+
+    with pytest.raises(RuntimeError, match="provision failed") as raised:
+        service.initialize(setup_input, initialization_validated=True)
+
+    assert raised.value is error
+    assert lock.exited
+    assert not lock.inside

packages/contracts/generated/api/console/agent/types.gen.ts

@@ -1882,6 +1882,8 @@ export type AgentKnowledgeMetadataCondition = {
     | '≠'
     | '≤'
     | '≥'
+  id?: string | null
+  metadata_id?: string | null
   name: string
   value?: string | Array<string> | number | null
 }

packages/contracts/generated/api/console/agent/zod.gen.ts

@@ -2357,6 +2357,14 @@ export const zAgentKnowledgeRetrievalConfig = z.object({
 
 /**
  * AgentKnowledgeMetadataCondition
+ *
+ * One manual metadata filter clause.
+ *
+ * ``id`` and ``metadata_id`` are UI-only bookkeeping the composer sends on
+ * every save (a stable row key and a reference to the selected metadata
+ * field). They are persisted here for round-tripping the composer's draft
+ * state but are stripped before building the Agent runtime request, whose
+ * DTO only accepts ``name``/``comparison_operator``/``value``.
  */
 export const zAgentKnowledgeMetadataCondition = z.object({
   comparison_operator: z.enum([
@@ -2379,6 +2387,8 @@ export const zAgentKnowledgeMetadataCondition = z.object({
     '≤',
     '≥',
   ]),
+  id: z.string().nullish(),
+  metadata_id: z.string().nullish(),
   name: z.string().min(1).max(255),
   value: z.union([z.string(), z.array(z.string()), z.number()]).nullish(),
 })

packages/contracts/generated/api/console/apps/types.gen.ts

@@ -3097,6 +3097,8 @@ export type AgentKnowledgeMetadataCondition = {
     | '≠'
     | '≤'
     | '≥'
+  id?: string | null
+  metadata_id?: string | null
   name: string
   value?: string | Array<string> | number | null
 }

packages/contracts/generated/api/console/apps/zod.gen.ts

@@ -4048,6 +4048,14 @@ export const zMessageInfiniteScrollPaginationResponse = z.object({
 
 /**
  * AgentKnowledgeMetadataCondition
+ *
+ * One manual metadata filter clause.
+ *
+ * ``id`` and ``metadata_id`` are UI-only bookkeeping the composer sends on
+ * every save (a stable row key and a reference to the selected metadata
+ * field). They are persisted here for round-tripping the composer's draft
+ * state but are stripped before building the Agent runtime request, whose
+ * DTO only accepts ``name``/``comparison_operator``/``value``.
  */
 export const zAgentKnowledgeMetadataCondition = z.object({
   comparison_operator: z.enum([
@@ -4070,6 +4078,8 @@ export const zAgentKnowledgeMetadataCondition = z.object({
     '≤',
     '≥',
   ]),
+  id: z.string().nullish(),
+  metadata_id: z.string().nullish(),
   name: z.string().min(1).max(255),
   value: z.union([z.string(), z.array(z.string()), z.number()]).nullish(),
 })

packages/contracts/generated/api/console/snippets/types.gen.ts

@@ -1130,6 +1130,8 @@ export type AgentKnowledgeMetadataCondition = {
     | '≠'
     | '≤'
     | '≥'
+  id?: string | null
+  metadata_id?: string | null
   name: string
   value?: string | Array<string> | number | null
 }

packages/contracts/generated/api/console/snippets/zod.gen.ts

@@ -1439,6 +1439,14 @@ export const zAgentKnowledgeRetrievalConfig = z.object({
 
 /**
  * AgentKnowledgeMetadataCondition
+ *
+ * One manual metadata filter clause.
+ *
+ * ``id`` and ``metadata_id`` are UI-only bookkeeping the composer sends on
+ * every save (a stable row key and a reference to the selected metadata
+ * field). They are persisted here for round-tripping the composer's draft
+ * state but are stripped before building the Agent runtime request, whose
+ * DTO only accepts ``name``/``comparison_operator``/``value``.
  */
 export const zAgentKnowledgeMetadataCondition = z.object({
   comparison_operator: z.enum([
@@ -1461,6 +1469,8 @@ export const zAgentKnowledgeMetadataCondition = z.object({
     '≤',
     '≥',
   ]),
+  id: z.string().nullish(),
+  metadata_id: z.string().nullish(),
   name: z.string().min(1).max(255),
   value: z.union([z.string(), z.array(z.string()), z.number()]).nullish(),
 })

web/app/(commonLayout)/(home)/loading.tsx

@@ -1,5 +0,0 @@
-import { HomeLoading } from '@/features/home/loading'
-
-export default function Loading() {
-  return <HomeLoading />
-}

web/app/components/base/features/new-feature-panel/annotation-reply/__tests__/annotation-ctrl-button.spec.tsx

@@ -60,7 +60,9 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    expect(screen.getByRole('button')).toBeInTheDocument()
+    expect(
+      screen.getByRole('button', { name: 'appDebug.feature.annotation.edit' }),
+    ).toBeInTheDocument()
   })
 
   it('should call onEdit when edit button is clicked', () => {
@@ -76,7 +78,7 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    fireEvent.click(screen.getByRole('button'))
+    fireEvent.click(screen.getByRole('button', { name: 'appDebug.feature.annotation.edit' }))
 
     expect(onEdit).toHaveBeenCalled()
   })
@@ -93,7 +95,9 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    expect(screen.getByRole('button')).toBeInTheDocument()
+    expect(
+      screen.getByRole('button', { name: 'appDebug.feature.annotation.add' }),
+    ).toBeInTheDocument()
   })
 
   it('should not render any button when not cached and no answer', () => {
@@ -125,7 +129,7 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    fireEvent.click(screen.getByRole('button'))
+    fireEvent.click(screen.getByRole('button', { name: 'appDebug.feature.annotation.add' }))
 
     await waitFor(() => {
       expect(mockAddAnnotation).toHaveBeenCalledWith('test-app', {
@@ -151,7 +155,7 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    fireEvent.click(screen.getByRole('button'))
+    fireEvent.click(screen.getByRole('button', { name: 'appDebug.feature.annotation.add' }))
 
     expect(mockSetShowAnnotationFullModal).toHaveBeenCalled()
     expect(mockAddAnnotation).not.toHaveBeenCalled()
@@ -176,7 +180,7 @@ describe('AnnotationCtrlButton', () => {
       />,
     )
 
-    fireEvent.click(screen.getByRole('button'))
+    fireEvent.click(screen.getByRole('button', { name: 'appDebug.feature.annotation.add' }))
 
     await waitFor(() => {
       expect(onAdded).toHaveBeenCalledWith('annotation-2', '')

web/app/components/base/features/new-feature-panel/annotation-reply/annotation-ctrl-button.tsx

@@ -1,9 +1,8 @@
 'use client'
-import type { FC } from 'react'
 import { toast } from '@langgenius/dify-ui/toast'
 import { Tooltip, TooltipContent, TooltipTrigger } from '@langgenius/dify-ui/tooltip'
 import { RiEditLine, RiFileEditLine } from '@remixicon/react'
-import * as React from 'react'
+import { memo } from 'react'
 import { useTranslation } from 'react-i18next'
 import ActionButton from '@/app/components/base/action-button'
 import { useModalContext } from '@/context/modal-context'
@@ -19,7 +18,7 @@ type Props = Readonly<{
   onAdded: (annotationId: string, authorName: string) => void
   onEdit: () => void
 }>
-const AnnotationCtrlButton: FC<Props> = ({
+const AnnotationCtrlButton = ({
   cached,
   query,
   answer,
@@ -27,7 +26,7 @@ const AnnotationCtrlButton: FC<Props> = ({
   messageId,
   onAdded,
   onEdit,
-}) => {
+}: Props) => {
   const { t } = useTranslation()
   const { plan, enableBilling } = useProviderContext()
   const isAnnotationFull =
@@ -52,8 +51,11 @@ const AnnotationCtrlButton: FC<Props> = ({
         <Tooltip>
           <TooltipTrigger
             render={
-              <ActionButton onClick={onEdit}>
-                <RiEditLine className="size-4" />
+              <ActionButton
+                aria-label={t(($) => $['feature.annotation.edit'], { ns: 'appDebug' })}
+                onClick={onEdit}
+              >
+                <RiEditLine aria-hidden className="size-4" />
               </ActionButton>
             }
           />
@@ -66,8 +68,11 @@ const AnnotationCtrlButton: FC<Props> = ({
         <Tooltip>
           <TooltipTrigger
             render={
-              <ActionButton onClick={handleAdd}>
-                <RiFileEditLine className="size-4" />
+              <ActionButton
+                aria-label={t(($) => $['feature.annotation.add'], { ns: 'appDebug' })}
+                onClick={handleAdd}
+              >
+                <RiFileEditLine aria-hidden className="size-4" />
               </ActionButton>
             }
           />
@@ -79,4 +84,4 @@ const AnnotationCtrlButton: FC<Props> = ({
     </>
   )
 }
-export default React.memo(AnnotationCtrlButton)
+export default memo(AnnotationCtrlButton)

web/features/home/loading.tsx

@@ -1,18 +0,0 @@
-'use client'
-
-import { useSuspenseQuery } from '@tanstack/react-query'
-import { systemFeaturesQueryOptions } from '@/features/system-features/client'
-import { HomeShell } from './home-shell'
-import { HomeSkeleton } from './home-skeleton'
-
-export function HomeLoading() {
-  const { data: systemFeatures } = useSuspenseQuery(systemFeaturesQueryOptions())
-
-  return (
-    <HomeShell>
-      <div className="flex flex-1 flex-col overflow-y-auto">
-        <HomeSkeleton showBanner={systemFeatures.enable_explore_banner} />
-      </div>
-    </HomeShell>
-  )
-}

web/features/home/page.tsx

@@ -1,4 +1,5 @@
 import { dehydrate, HydrationBoundary } from '@tanstack/react-query'
+import { Suspense } from 'react'
 import { getQueryClient } from '@/app/get-query-client'
 import {
   getSystemFeaturesQueryClient,
@@ -7,6 +8,8 @@ import {
 import { getLocaleOnServer } from '@/i18n-config/server'
 import { getServerConsoleClientContext, serverConsoleQuery } from '@/service/server'
 import { HomeContent } from './home-content/home-content'
+import { HomeShell } from './home-shell'
+import { HomeSkeleton } from './home-skeleton'
 
 export async function HomePage() {
   const homeQueryClient = getQueryClient()
@@ -44,7 +47,17 @@ export async function HomePage() {
 
   return (
     <HydrationBoundary state={dehydratedState}>
-      <HomeContent />
+      <Suspense
+        fallback={
+          <HomeShell>
+            <div className="flex flex-1 flex-col overflow-y-auto">
+              <HomeSkeleton showBanner={enableExploreBanner} />
+            </div>
+          </HomeShell>
+        }
+      >
+        <HomeContent />
+      </Suspense>
     </HydrationBoundary>
   )
 }