refactor(api): switch application services and execution entry points

.github/workflows/knowledge-fs-ci.yml

@@ -95,7 +95,7 @@ jobs:
               - 'api/tests/unit_tests/core/workflow/test_node_runtime.py'
               - 'api/tests/unit_tests/services/enterprise/test_rbac_service.py'
               - 'api/tests/unit_tests/repositories/workspace/test_workspace_repository.py'
-              - 'api/tests/unit_tests/services/test_agent_tool_inner_service.py'
+              - 'api/tests/unit_tests/services/agent/test_tool_invocation_service.py'
               - 'api/tests/unit_tests/services/test_vector_space_admission_service.py'
               - 'api/tests/unit_tests/services/test_knowledge_fs_capability.py'
               - 'api/tests/unit_tests/services/knowledge_fs/test_product_operations.py'
@@ -275,7 +275,7 @@ jobs:
             api/tests/unit_tests/core/workflow/test_structured_output_compatibility.py
             api/tests/unit_tests/services/enterprise/test_rbac_service.py
             api/tests/unit_tests/repositories/workspace/test_workspace_repository.py
-            api/tests/unit_tests/services/test_agent_tool_inner_service.py
+            api/tests/unit_tests/services/agent/test_tool_invocation_service.py
             api/tests/unit_tests/services/test_vector_space_admission_service.py
             api/tests/unit_tests/services/test_tidb_binding_service.py
           )

api/.importlinter

@@ -32,6 +32,8 @@ layers =
 # error makes this list shrink whenever a legacy dependency is removed.
 unmatched_ignore_imports_alerting = error
 ignore_imports =
+    core.app.apps.advanced_chat.app_runner -> extensions.workflow_warm_shutdown
+    core.app.apps.workflow.app_runner -> extensions.workflow_warm_shutdown
     # The Dify node factory composes the Human Input runtime for workflow execution.
     core.workflow.node_factory -> services.human_input_v2.composition
     # Legacy generators delegate configured file defaults to the application owner during core migration.
@@ -76,7 +78,6 @@ ignore_imports =
     core.workflow.workflow_entry -> services.workflow_node_variables
     core.app.apps.workflow_app_runner -> tasks.mail_human_input_delivery_task
     core.app.features.annotation_reply.annotation_reply -> services.annotation_service
-    core.app.features.annotation_reply.annotation_reply -> services.knowledge.dataset_service
     core.app.llm.quota -> services.credit_pool_service
     core.app.task_pipeline.message_cycle_manager -> services.annotation_service
     core.helper.credential_utils -> services.enterprise.plugin_manager_service
@@ -105,7 +106,6 @@ ignore_imports =
     # The KnowledgeFS retrieval node is owned by services.knowledge_fs and only wired in here.
     core.workflow.node_factory -> services.knowledge_fs.knowledge_retrieval_v2.automatic_metadata_filter
     core.workflow.node_factory -> services.knowledge_fs.knowledge_retrieval_v2.entities
-    core.workflow.nodes.agent.message_transformer -> services.tools.legacy_builtin_tools_manage_service
     core.workflow.nodes.agent_v2.agent_node -> services.agent.knowledge_citations
     core.workflow.nodes.agent_v2.agent_node -> services.agent.prompt_mentions
     core.workflow.nodes.agent_v2.agent_node -> services.agent.workspace_service
@@ -134,7 +134,6 @@ ignore_imports =
     libs.oauth_bearer -> models
     libs.rsa -> extensions.ext_storage
     libs.workspace_permission -> services.enterprise.enterprise_service
-    services.app_generate_service -> controllers.console.app.workflow
 
 [importlinter:contract:no-direct-rsa-imports]
 # Note: `libs` itself is deliberately excluded from source_modules -- import-linter's

api/celery_healthcheck.py

@@ -1,18 +1,18 @@
-# This module provides a lightweight Celery instance for use in Docker health checks.
-# Unlike celery_entrypoint.py, this does NOT import app.py and therefore avoids
-# initializing all Flask extensions (DB, Redis, storage, blueprints, etc.).
-# Using this module keeps the health check fast and low-cost.
-from celery import Celery
-
-from configs import dify_config
-from extensions.ext_celery import get_celery_broker_transport_options, get_celery_ssl_options
-
-celery = Celery(broker=dify_config.CELERY_BROKER_URL)
-
-broker_transport_options = get_celery_broker_transport_options()
-if broker_transport_options:
-    celery.conf.update(broker_transport_options=broker_transport_options)
-
-ssl_options = get_celery_ssl_options()
-if ssl_options:
-    celery.conf.update(broker_use_ssl=ssl_options)
+# This module provides a lightweight Celery instance for use in Docker health checks.
+# Unlike celery_entrypoint.py, this does NOT import app.py and therefore avoids
+# initializing all Flask extensions (DB, Redis, storage, blueprints, etc.).
+# Using this module keeps the health check fast and low-cost.
+from celery import Celery
+
+from configs import dify_config
+from extensions.ext_celery import get_celery_broker_transport_options, get_celery_ssl_options
+
+celery = Celery(broker=dify_config.CELERY_BROKER_URL)
+
+broker_transport_options = get_celery_broker_transport_options()
+if broker_transport_options:
+    celery.conf.update(broker_transport_options=broker_transport_options)
+
+ssl_options = get_celery_ssl_options()
+if ssl_options:
+    celery.conf.update(broker_use_ssl=ssl_options)

api/commands/data_migration.py

@@ -23,6 +23,7 @@
     migrate_resource_whitelist_scopes_to_automatic_include,  # noqa: F401
 )
 from core.db.session_factory import session_factory
+from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import ModelType
 from models import Tenant
@@ -290,7 +291,10 @@ def export_migration_data(input_file: str | None, output_file: str | None, overw
         raw_config = _load_json_object(input_file, "Export config")
         selection = ExportConfigParser().parse(raw_config)
         with session_factory.create_session() as session:
-            result = MigrationExportService().export(selection, session=session)
+            result = MigrationExportService(
+                workflow_tools=application_services().tools.workflows,
+                tool_providers=application_services().tools.tool_providers,
+            ).export(selection, session=session)
         MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
         click.echo(click.style(f"Output written to {output_file}", fg="green"))
         _render_report(result.report_items, context=_with_output_path(result.report_context, output_file))
@@ -338,7 +342,13 @@ def import_migration_data(
         assert input_file is not None
         package = MigrationPackageService().load_package(input_file)
         with session_factory.create_session() as session:
-            result = MigrationImportService().import_package(
+            from extensions.application_services.workflow import build_app_dsl_service
+
+            result = MigrationImportService(
+                workflow_tools=application_services().tools.workflows,
+                workflows=application_services().console_workflows,
+                app_dsl=build_app_dsl_service,
+            ).import_package(
                 ImportRequest(
                     package=package,
                     cli_target_tenant=target_tenant,
@@ -437,7 +447,10 @@ def migration_data_wizard() -> None:
             output_file=output_file,
         )
         with session_factory.create_session() as session:
-            result = MigrationExportService().export(selection, session=session)
+            result = MigrationExportService(
+                workflow_tools=application_services().tools.workflows,
+                tool_providers=application_services().tools.tool_providers,
+            ).export(selection, session=session)
         MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
         click.echo(click.style(f"Output written to {output_file}", fg="green"))
         _print_wizard_step("Report")

api/controllers/console/agent/composer.py

@@ -2,8 +2,8 @@
 
 from flask_restx import Resource
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound
 
+from controllers.common.errors import NotFoundError
 from controllers.common.rbac import AgentId, PlainApp, RBACCheck, Workspace
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.common.session import with_session
@@ -19,6 +19,8 @@
     with_current_tenant_id,
     with_current_user_id,
 )
+from extensions.application_services.snippets import build_snippet_service
+from extensions.ext_application_services import application_services
 from fields.agent_fields import (
     AgentAppComposerResponse,
     AgentComposerCandidatesResponse,
@@ -46,7 +48,6 @@
     WorkflowAgentComposerQuery,
     WorkflowComposerCopyFromRosterPayload,
 )
-from services.snippet_service import SnippetService
 
 register_schema_models(
     console_ns, ComposerSavePayload, WorkflowAgentComposerQuery, WorkflowComposerCopyFromRosterPayload
@@ -225,11 +226,13 @@ def get(self, session: Session, tenant_id: str, current_user_id: str, app_model:
         return dump_response(
             AgentComposerCandidatesResponse,
             AgentComposerService.get_workflow_candidates(
+                variables=application_services().workflow_variables,
                 session=session,
                 tenant_id=tenant_id,
                 app_id=app_model.id,
                 node_id=node_id,
                 user_id=current_user_id,
+                tool_providers=application_services().tools.tool_providers,
             ),
         )
 
@@ -307,12 +310,12 @@ def post(
 
 
 def _require_snippet_app_id(*, session: Session, tenant_id: str, snippet_id: UUID) -> str:
-    snippet = SnippetService(session=session).get_snippet_by_id(
+    snippet = build_snippet_service(session=session).get_snippet_by_id(
         snippet_id=str(snippet_id),  # pyrefly: ignore[unnecessary-type-conversion]
         tenant_id=tenant_id,
     )
     if snippet is None:
-        raise NotFound("Snippet not found")
+        raise NotFoundError("Snippet not found")
     return snippet.id
 
 
@@ -470,11 +473,13 @@ def get(self, session: Session, tenant_id: str, current_user_id: str, snippet_id
         return dump_response(
             AgentComposerCandidatesResponse,
             AgentComposerService.get_workflow_candidates(
+                variables=application_services().workflow_variables,
                 session=session,
                 tenant_id=tenant_id,
                 app_id=_require_snippet_app_id(session=session, tenant_id=tenant_id, snippet_id=snippet_id),
                 node_id=node_id,
                 user_id=current_user_id,
+                tool_providers=application_services().tools.tool_providers,
             ),
         )
 
@@ -650,5 +655,6 @@ def get(self, session: Session, tenant_id: str, current_user_id: str, agent_id:
                 tenant_id=tenant_id,
                 agent_id=str(agent_id),
                 user_id=current_user_id,
+                tool_providers=application_services().tools.tool_providers,
             ),
         )

api/controllers/console/agent/roster.py

@@ -1,14 +1,15 @@
 from typing import Literal
 from uuid import UUID
 
-from flask import abort, request
+from flask import request
 from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, field_validator
 from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from configs import dify_config
 from controllers.common.agent_access import resolve_agent_access_filter
+from controllers.common.errors import AccessDeniedError, InvalidRequestError
 from controllers.common.rbac import AgentId, RBACCheck, Workspace
 from controllers.common.schema import (
     query_params_from_model,
@@ -431,7 +432,14 @@ def _serialize_agent_app_detail(
 
     roster_service = _agent_roster_service(session)
     payload = GenericAppDetailWithSite.model_validate(
-        AppResponseView(app_model, session=session, account=current_user, access_mode=access_mode),
+        AppResponseView(
+            app_model,
+            tool_providers=application_services().tools.tool_providers,
+            workflow_queries=application_services().tools.workflow_queries,
+            session=session,
+            account=current_user,
+            access_mode=access_mode,
+        ),
         from_attributes=True,
     ).model_dump(mode="json")
     agent = (
@@ -524,7 +532,15 @@ def _serialize_agent_app_pagination(
             "limit": app_pagination.per_page,
             "total": app_pagination.total,
             "has_more": app_pagination.has_next,
-            "data": [AppResponseView(app, session=session) for app in app_pagination.items],
+            "data": [
+                AppResponseView(
+                    app,
+                    tool_providers=application_services().tools.tool_providers,
+                    workflow_queries=application_services().tools.workflow_queries,
+                    session=session,
+                )
+                for app in app_pagination.items
+            ],
         },
     ).model_dump(mode="json")
     payload["publication_counts"] = {
@@ -623,7 +639,7 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:
     try:
         return parse_time_range(start, end, timezone)
     except ValueError as exc:
-        abort(400, description=str(exc))
+        raise InvalidRequestError(str(exc)) from exc
 
 
 def _query_values(name: str, alias_name: str | None = None) -> list[str]:
@@ -1210,7 +1226,7 @@ def get(self, session: Session, tenant_id: str, current_user: Account, agent_id:
                 ),
             )
         except ValueError as exc:
-            abort(400, description=str(exc))
+            raise InvalidRequestError(str(exc)) from exc
         return dump_response(AgentLogListResponse, payload)
 
 
@@ -1251,7 +1267,7 @@ def get(self, session: Session, tenant_id: str, current_user: Account, agent_id:
                 ),
             )
         except ValueError as exc:
-            abort(400, description=str(exc))
+            raise InvalidRequestError(str(exc)) from exc
         return dump_response(AgentLogMessageListResponse, payload)
 
 
@@ -1307,7 +1323,7 @@ def get(
                 params=AgentStatisticsQueryParams(source=req_data.source, start=start, end=end, timezone=timezone),
             )
         except ValueError as exc:
-            abort(400, description=str(exc))
+            raise InvalidRequestError(str(exc)) from exc
         return dump_response(AgentStatisticSummaryEnvelopeResponse, payload)
 
 
@@ -1366,7 +1382,7 @@ def post(self, session: Session, tenant_id: str, current_user: Account, agent_id
             dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD
             and not FeatureService.get_workspace_plan(tenant_id).is_paid
         ):
-            abort(403, description="This feature requires a paid plan.")
+            raise AccessDeniedError("This feature requires a paid plan.")
 
         return dump_response(
             AgentConfigSnapshotRestoreResponse,

api/controllers/console/app/completion.py

@@ -8,9 +8,9 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 import services
+from controllers.common.errors import InternalServerError, InvalidRequestError, NotFoundError
 from controllers.common.fields import SimpleResultResponse
 from controllers.common.rbac import AgentId, PlainApp, RBACCheck
 from controllers.common.schema import register_response_schema_models, register_schema_models
@@ -45,6 +45,7 @@
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
+from extensions.ext_application_services import application_services
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.external_api import error_responses
@@ -74,7 +75,7 @@ def _resolve_debugger_chat_streaming(
     if app_mode != AppMode.AGENT:
         return response_mode != "blocking"
     if response_mode_provided and response_mode == "blocking":
-        raise BadRequest("Agent App only supports streaming response mode.")
+        raise InvalidRequestError("Agent App only supports streaming response mode.")
     return True
 
 
@@ -178,6 +179,8 @@ def post(self, req_data: CompletionMessagePayload, session: Session, current_use
 
         try:
             response = AppGenerateService.generate(
+                variables=application_services().workflow_variables,
+                runtime=application_services().workflow_runtime,
                 session=session,
                 app_model=app_model,
                 user=current_user,
@@ -189,7 +192,7 @@ def post(self, req_data: CompletionMessagePayload, session: Session, current_use
             # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)
         except services.errors.conversation.ConversationNotExistsError as caught_error:
-            raise NotFound("Conversation Not Exists.") from caught_error
+            raise NotFoundError("Conversation Not Exists.") from caught_error
         except services.errors.conversation.ConversationCompletedError as caught_error:
             raise ConversationCompletedError() from caught_error
         except services.errors.app_model_config.AppModelConfigBrokenError as caught_error:
@@ -420,7 +423,7 @@ def _resolve_current_user_agent_debug_conversation_id(
         account_id=current_user.id,
     )
     if conversation_id is None:
-        raise NotFound("Conversation Not Exists.")
+        raise NotFoundError("Conversation Not Exists.")
     return conversation_id
 
 
@@ -450,7 +453,7 @@ def _create_chat_message(
             start_new=draft_type == AgentConfigDraftType.DRAFT and not args_model.conversation_id,
         )
         if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
-            raise NotFound("Conversation Not Exists.")
+            raise NotFoundError("Conversation Not Exists.")
         args["conversation_id"] = debug_conversation_id
 
     streaming = _resolve_debugger_chat_streaming(
@@ -562,6 +565,8 @@ def _generate_chat_message(
 ):
     try:
         return AppGenerateService.generate(
+            variables=application_services().workflow_variables,
+            runtime=application_services().workflow_runtime,
             session=session,
             app_model=app_model,
             user=current_user,
@@ -570,7 +575,7 @@ def _generate_chat_message(
             streaming=streaming,
         )
     except services.errors.conversation.ConversationNotExistsError as caught_error:
-        raise NotFound("Conversation Not Exists.") from caught_error
+        raise NotFoundError("Conversation Not Exists.") from caught_error
     except services.errors.conversation.ConversationCompletedError as caught_error:
         raise ConversationCompletedError() from caught_error
     except services.errors.app_model_config.AppModelConfigBrokenError as caught_error: