refactor(api): migrate Human Input forms and Workflow event streams

api/controllers/console/human_input_form.py

@@ -27,11 +27,6 @@
     with_current_tenant_id,
     with_current_user,
 )
-from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
-from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
-from core.app.apps.message_generator import MessageGenerator
-from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
 from extensions.ext_database import db
 from libs.external_api import error_responses
@@ -42,13 +37,16 @@
 from models.model import AppMode
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
+from services.app.generation.response import convert_to_event_stream
 from services.human_input_service import (
     Form,
     FormExpiredError,
     FormSubmittedError,
     HumanInputService,
     InvalidFormDataError,
 )
+from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter
+from services.workflow.execution.adapters.response_stream import WorkflowEventStream
 from services.workflow_event_snapshot_service import build_workflow_event_stream
 
 logger = logging.getLogger(__name__)
@@ -218,15 +216,9 @@ def _generate_finished_events() -> Generator[str]:
             event_generator = _generate_finished_events
 
         else:
-            msg_generator = MessageGenerator()
-            generator: type[BaseAppGenerator]
-            match app.mode:
-                case AppMode.ADVANCED_CHAT:
-                    generator = AdvancedChatAppGenerator
-                case AppMode.WORKFLOW:
-                    generator = WorkflowAppGenerator
-                case _:
-                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+            msg_generator = WorkflowEventStream
+            if app.mode not in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
             continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"
@@ -247,7 +239,7 @@ def _generate_finished_events() -> Generator[str]:
 
             def _generate_stream_events():
                 if include_state_snapshot:
-                    return generator.convert_to_event_stream(
+                    return convert_to_event_stream(
                         build_workflow_event_stream(
                             app_mode=AppMode(app.mode),
                             workflow_run=workflow_run,
@@ -259,7 +251,7 @@ def _generate_stream_events():
                             close_on_pause=not continue_on_pause,
                         )
                     )
-                return generator.convert_to_event_stream(
+                return convert_to_event_stream(
                     msg_generator.retrieve_events(AppMode(app.mode), workflow_run.id),
                 )
 

api/controllers/openapi/workflow_events.py

@@ -13,9 +13,9 @@
 from flask import Response
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from werkzeug.exceptions import NotFound, UnprocessableEntity
 
 from constants.oauth_bearer import Scope
+from controllers.common.errors import NotFoundError, UnprocessableEntityError
 from controllers.common.fields import EventStreamResponse
 from controllers.common.rbac import PlainApp, RBACCheck, RBACPermission
 from controllers.openapi import openapi_ns
@@ -32,16 +32,14 @@
 )
 from controllers.openapi.auth.subjects import AccountSubject, ExternalSsoSubject
 from controllers.openapi.human_input_form import with_form_hints
-from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
-from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
-from core.app.apps.message_generator import MessageGenerator
-from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.task_entities import StreamEvent
 from core.db.session_factory import session_factory
 from core.workflow.human_input_policy import HumanInputSurface
 from models.model import AppMode
 from repositories.factory import DifyAPIRepositoryFactory
+from services.app.generation.response import convert_to_event_stream
+from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter
+from services.workflow.execution.adapters.response_stream import WorkflowEventStream
 from services.workflow_event_snapshot_service import build_workflow_event_stream
 
 
@@ -86,7 +84,7 @@ def get(self, ctx: Context, app_id: str, run_id: str, *, query: WorkflowEventsQu
 
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.WORKFLOW, AppMode.ADVANCED_CHAT}:
-            raise UnprocessableEntity("mode_not_supported_for_event_reconnect")
+            raise UnprocessableEntityError("mode_not_supported_for_event_reconnect")
 
         # The event stream outlives `ctx.session`, so this route needs a maker of its
         # own — the guard's, not a fresh one bound straight to the engine.
@@ -98,15 +96,15 @@ def get(self, ctx: Context, app_id: str, run_id: str, *, query: WorkflowEventsQu
         )
 
         if workflow_run is None:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         if workflow_run.app_id != app_model.id:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         # Ownership is a property of the run row, not of the app, so no pipeline
         # requirement can answer it — the caller may only reconnect to its own run.
         if workflow_run.created_by_role != ctx.subject.caller_role or workflow_run.created_by != caller.id:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         workflow_run_entity = workflow_run
         tenant_id = app_model.tenant_id
@@ -126,17 +124,14 @@ def _generate_finished_events() -> Generator[str]:
 
             event_generator = _generate_finished_events
         else:
-            msg_generator = MessageGenerator()
-            generator: type[BaseAppGenerator]
-            generator = AdvancedChatAppGenerator if app_mode == AppMode.ADVANCED_CHAT else WorkflowAppGenerator
-
+            msg_generator = WorkflowEventStream
             include_state_snapshot = query.include_state_snapshot
             continue_on_pause = query.continue_on_pause
             terminal_events: list[StreamEvent] | None = [] if continue_on_pause else None
 
             def _generate_stream_events():
                 if include_state_snapshot:
-                    return generator.convert_to_event_stream(
+                    return convert_to_event_stream(
                         build_workflow_event_stream(
                             app_mode=app_mode,
                             workflow_run=workflow_run_entity,
@@ -147,7 +142,7 @@ def _generate_stream_events():
                             close_on_pause=not continue_on_pause,
                         )
                     )
-                return generator.convert_to_event_stream(
+                return convert_to_event_stream(
                     msg_generator.retrieve_events(
                         app_mode,
                         workflow_run_entity.id,

api/controllers/service_api/app/human_input_form.py

@@ -15,6 +15,7 @@
 from pydantic import ConfigDict, WithJsonSchema
 from werkzeug.exceptions import BadRequest, NotFound
 
+from controllers.common.errors import InvalidRequestError, NotFoundError
 from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console.wraps import model_validate
@@ -87,14 +88,14 @@ def _jsonify_form_definition(form: Form, *, inputs: Sequence[FormInputConfig] =
 
 def _ensure_form_belongs_to_app(form: Form, app_model: App) -> None:
     if form.app_id != app_model.id or form.tenant_id != app_model.tenant_id:
-        raise NotFound("Form not found")
+        raise NotFoundError("Form not found")
 
 
 def _ensure_form_is_allowed_for_service_api(form: Form) -> None:
     # Keep app-token callers scoped to the public web-form surface; internal HITL
     # routes must continue to flow through console-only authentication.
     if not is_recipient_type_allowed_for_surface(form.recipient_type, HumanInputSurface.SERVICE_API):
-        raise NotFound("Form not found")
+        raise NotFoundError("Form not found")
 
 
 def _jsonify_v2_form_definition(form: V2Form) -> Response:
@@ -252,7 +253,7 @@ def post(self, payload: HumanInputFormSubmitPayload, app_model: App, end_user: E
         recipient_type = form.recipient_type
         if recipient_type is None:
             logger.warning("Recipient type is None for form, form_id=%s", form.id)
-            raise BadRequest("Form recipient type is invalid")
+            raise InvalidRequestError("Form recipient type is invalid")
 
         try:
             service.submit_form_by_token(
@@ -263,6 +264,6 @@ def post(self, payload: HumanInputFormSubmitPayload, app_model: App, end_user: E
                 submission_end_user_id=end_user.id,
             )
         except FormNotFoundError as caught_error:
-            raise NotFound("Form not found") from caught_error
+            raise NotFoundError("Form not found") from caught_error
 
         return {}, 200

api/controllers/service_api/app/workflow_events.py

@@ -9,26 +9,24 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import sessionmaker
-from werkzeug.exceptions import NotFound
 
+from controllers.common.errors import NotFoundError
 from controllers.common.fields import EventStreamResponse
 from controllers.common.schema import query_params_from_model, register_response_schema_model, register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import NotWorkflowAppError
 from controllers.service_api.schema import event_stream_response
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
-from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
-from core.app.apps.message_generator import MessageGenerator
-from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.task_entities import StreamEvent
 from core.workflow.human_input_policy import HumanInputSurface
 from extensions.ext_database import db
 from libs.external_api import error_responses
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, EndUser
 from repositories.factory import DifyAPIRepositoryFactory
+from services.app.generation.response import convert_to_event_stream
+from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter
+from services.workflow.execution.adapters.response_stream import WorkflowEventStream
 from services.workflow_event_snapshot_service import build_workflow_event_stream
 
 
@@ -107,16 +105,16 @@ def get(self, app_model: App, end_user: EndUser, workflow_run_id: str):
         )
 
         if workflow_run is None:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         if workflow_run.app_id != app_model.id:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         if workflow_run.created_by_role != CreatorUserRole.END_USER:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         if workflow_run.created_by != end_user.id:
-            raise NotFound("Workflow run not found")
+            raise NotFoundError("Workflow run not found")
 
         workflow_run_entity = workflow_run
 
@@ -135,13 +133,8 @@ def _generate_finished_events() -> Generator[str]:
 
             event_generator = _generate_finished_events
         else:
-            msg_generator = MessageGenerator()
-            generator: type[BaseAppGenerator]
-            if app_mode == AppMode.ADVANCED_CHAT:
-                generator = AdvancedChatAppGenerator
-            elif app_mode == AppMode.WORKFLOW:
-                generator = WorkflowAppGenerator
-            else:
+            msg_generator = WorkflowEventStream
+            if app_mode not in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
                 raise NotWorkflowAppError()
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
@@ -150,7 +143,7 @@ def _generate_finished_events() -> Generator[str]:
 
             def _generate_stream_events():
                 if include_state_snapshot:
-                    return generator.convert_to_event_stream(
+                    return convert_to_event_stream(
                         build_workflow_event_stream(
                             app_mode=app_mode,
                             workflow_run=workflow_run_entity,
@@ -161,7 +154,7 @@ def _generate_stream_events():
                             close_on_pause=not continue_on_pause,
                         )
                     )
-                return generator.convert_to_event_stream(
+                return convert_to_event_stream(
                     msg_generator.retrieve_events(
                         app_mode,
                         workflow_run_entity.id,

api/controllers/web/human_input_form.py

@@ -9,10 +9,9 @@
 from flask import request
 from flask_restx import Resource
 from sqlalchemy import select
-from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.errors import NotFoundError
+from controllers.common.errors import AccessDeniedError, NotFoundError
 from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console.wraps import model_validate
@@ -162,7 +161,7 @@ class HumanInputFormApi(Resource):
     @web_ns.doc(
         responses={
             200: "Form retrieved successfully",
-            403: "Forbidden",
+            403: "AccessDeniedError",
             404: "Form not found",
             412: "Form already submitted or expired",
             429: "Too many requests",
@@ -191,7 +190,7 @@ def get(self, form_token: str):
         app_model, site = _get_app_site_from_form(form)
         tenant = app_model.tenant(db.session)
         if tenant is None:
-            raise Forbidden()
+            raise AccessDeniedError()
         inputs = service.resolve_form_inputs(form)
         features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
 
@@ -296,10 +295,10 @@ def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
 
     site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
     if site is None:
-        raise Forbidden()
+        raise AccessDeniedError()
 
     tenant = app_model.tenant(db.session)
     if tenant is None or tenant.status == TenantStatus.ARCHIVE:
-        raise Forbidden()
+        raise AccessDeniedError()
 
     return app_model, site

api/controllers/web/workflow_events.py

@@ -13,16 +13,14 @@
 from controllers.common.schema import register_response_schema_model
 from controllers.web import api, web_ns
 from controllers.web.wraps import WebApiResource
-from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
-from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
-from core.app.apps.message_generator import MessageGenerator
-from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from extensions.ext_database import db
 from libs.external_api import error_responses
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, EndUser
 from repositories.factory import DifyAPIRepositoryFactory
+from services.app.generation.response import convert_to_event_stream
+from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter
+from services.workflow.execution.adapters.response_stream import WorkflowEventStream
 from services.workflow_event_snapshot_service import build_workflow_event_stream
 
 register_response_schema_model(web_ns, EventStreamResponse)
@@ -77,22 +75,16 @@ def _generate_finished_events() -> Generator[str]:
             event_generator = _generate_finished_events
         else:
             app_mode = AppMode.value_of(app_model.mode)
-            msg_generator = MessageGenerator()
-            generator: type[BaseAppGenerator]
-            match app_mode:
-                case AppMode.ADVANCED_CHAT:
-                    generator = AdvancedChatAppGenerator
-                case AppMode.WORKFLOW:
-                    generator = WorkflowAppGenerator
-                case _:
-                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+            msg_generator = WorkflowEventStream
+            if app_mode not in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
             continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"
 
             def _generate_stream_events():
                 if include_state_snapshot:
-                    return generator.convert_to_event_stream(
+                    return convert_to_event_stream(
                         build_workflow_event_stream(
                             app_mode=app_mode,
                             workflow_run=workflow_run,
@@ -102,7 +94,7 @@ def _generate_stream_events():
                             close_on_pause=not continue_on_pause,
                         )
                     )
-                return generator.convert_to_event_stream(
+                return convert_to_event_stream(
                     msg_generator.retrieve_events(app_mode, workflow_run.id),
                 )