diff --git a/api/controllers/console/human_input_form.py b/api/controllers/console/human_input_form.py index 4613228421121a..9d0ed77955063c 100644 --- a/api/controllers/console/human_input_form.py +++ b/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,6 +37,7 @@ 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, @@ -49,6 +45,8 @@ 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), ) diff --git a/api/controllers/openapi/workflow_events.py b/api/controllers/openapi/workflow_events.py index 6b35be8a9fddba..a14b6155a397e4 100644 --- a/api/controllers/openapi/workflow_events.py +++ b/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, diff --git a/api/controllers/service_api/app/human_input_form.py b/api/controllers/service_api/app/human_input_form.py index e398c46aa25811..fa8f09910c9456 100644 --- a/api/controllers/service_api/app/human_input_form.py +++ b/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 diff --git a/api/controllers/service_api/app/workflow_events.py b/api/controllers/service_api/app/workflow_events.py index 1aaf2bbf122db3..90b3f3d0e7ea7b 100644 --- a/api/controllers/service_api/app/workflow_events.py +++ b/api/controllers/service_api/app/workflow_events.py @@ -9,19 +9,14 @@ 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 @@ -29,6 +24,9 @@ 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, diff --git a/api/controllers/web/human_input_form.py b/api/controllers/web/human_input_form.py index a1c6fa4e25419a..7acc9e71e2b415 100644 --- a/api/controllers/web/human_input_form.py +++ b/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 diff --git a/api/controllers/web/workflow_events.py b/api/controllers/web/workflow_events.py index d74ebcfcdd553a..b6c045377c6b5c 100644 --- a/api/controllers/web/workflow_events.py +++ b/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), ) diff --git a/api/openapi/markdown/web-openapi.md b/api/openapi/markdown/web-openapi.md index 3926fc2d2286e9..a28a23c52c96f3 100644 --- a/api/openapi/markdown/web-openapi.md +++ b/api/openapi/markdown/web-openapi.md @@ -501,7 +501,7 @@ GET /api/form/human_input/ | Code | Description | Schema | | ---- | ----------- | ------ | | 200 | Form retrieved successfully | **application/json**: [HumanInputFormDefinitionResponse](#humaninputformdefinitionresponse)
| -| 403 | Forbidden | **application/json**: [ErrorResponse](#errorresponse)
| +| 403 | AccessDeniedError | **application/json**: [ErrorResponse](#errorresponse)
| | 404 | Form not found | **application/json**: [ErrorResponse](#errorresponse)
| | 412 | Form already submitted or expired | **application/json**: [HumanInputFormExpiredError](#humaninputformexpirederror)
[HumanInputFormSubmittedError](#humaninputformsubmittederror)
[ErrorResponse](#errorresponse)
| | 429 | Too many requests | **application/json**: [WebFormRateLimitExceededError](#webformratelimitexceedederror)
[ErrorResponse](#errorresponse)
| diff --git a/api/services/human_input_service.py b/api/services/human_input_service.py index 6b1cd1f0265cf8..7f34c4b776f80a 100644 --- a/api/services/human_input_service.py +++ b/api/services/human_input_service.py @@ -10,7 +10,6 @@ from configs import dify_config from core.app.file_access import DatabaseFileAccessController from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext -from core.repositories.human_input_repository import HumanInputFormSubmissionRepository from core.workflow.human_input_policy import resolve_variable_select_input_options from enums.human_input import HumanInputFormKind, HumanInputFormStatus, RecipientType, ValueSourceType from factories.file_factory import build_from_mapping, build_from_mappings @@ -31,9 +30,12 @@ SelectInputConfig, UserActionConfig, ) -from models.human_input_entities import validate_human_input_submission as graphon_validate_human_input_submission +from models.human_input_entities import ( + validate_human_input_submission as graphon_validate_human_input_submission, +) from models.model import App, AppMode from repositories.factory import DifyAPIRepositoryFactory +from repositories.human_input.form_repository import HumanInputFormSubmissionRepository from repositories.human_input_errors import FormAlreadyHandledError from tasks.app_generate.workflow_execute_task import resume_app_execution @@ -161,7 +163,7 @@ def __init__( if isinstance(session_factory, Engine): session_factory = sessionmaker(bind=session_factory) self._session_factory = session_factory - self._form_repository = form_repository or HumanInputFormSubmissionRepository() + self._form_repository = form_repository or HumanInputFormSubmissionRepository(sessions=self._session_factory) def get_form_by_token(self, form_token: str) -> Form | None: record = self._form_repository.get_by_token(form_token) @@ -336,9 +338,8 @@ def _is_globally_expired(self, form: Form, *, now: datetime | None = None) -> bo global_deadline = created_at + timedelta(seconds=global_timeout_seconds) return global_deadline <= current - @classmethod def validate_and_normalize_submission( - cls, + self, *, tenant_id: str, form_definition: FormDefinitionProtocol, @@ -351,7 +352,7 @@ def validate_and_normalize_submission( graphon owns the form schema and validation rules, while Dify owns tenant-aware file reconstruction and persistence compatibility for submitted payloads. """ - normalized_form_data = cls.normalize_submission_data( + normalized_form_data = self.normalize_submission_data( tenant_id=tenant_id, form_definition=form_definition, form_data=form_data, @@ -364,9 +365,8 @@ def validate_and_normalize_submission( ) return normalized_form_data - @classmethod def normalize_submission_data( - cls, + self, *, tenant_id: str, form_definition: FormDefinitionProtocol, @@ -377,7 +377,7 @@ def normalize_submission_data( for name, form_input in inputs_by_name.items(): if name not in form_data: continue - normalized_form_data[name] = cls._normalize_input_value( + normalized_form_data[name] = self._normalize_input_value( tenant_id=tenant_id, form_input=form_input, value=form_data[name], @@ -385,9 +385,8 @@ def normalize_submission_data( return normalized_form_data - @classmethod def _normalize_input_value( - cls, + self, *, tenant_id: str, form_input: FormInputConfig, @@ -400,15 +399,15 @@ def _normalize_input_value( ) return value if isinstance(form_input, SelectInputConfig): - return cls._normalize_select_value(form_input=form_input, value=value) + return self._normalize_select_value(form_input=form_input, value=value) if isinstance(form_input, FileInputConfig): - return cls._normalize_file_value( + return self._normalize_file_value( tenant_id=tenant_id, form_input=form_input, value=value, ) if isinstance(form_input, FileListInputConfig): - return cls._normalize_file_list_value( + return self._normalize_file_list_value( tenant_id=tenant_id, form_input=form_input, value=value, @@ -433,9 +432,8 @@ def _normalize_select_value( ) return value - @classmethod def _normalize_file_value( - cls, + self, *, tenant_id: str, form_input: FileInputConfig, @@ -445,7 +443,7 @@ def _normalize_file_value( raise HumanInputSubmissionValidationError( f"Invalid value for file input '{form_input.output_variable_name}': expected mapping" ) - upload_config = cls._build_file_upload_config(form_input=form_input, number_limits=1) + upload_config = self._build_file_upload_config(form_input=form_input, number_limits=1) try: # `build_from_mapping` enforces tenant ownership for persisted upload references. file = build_from_mapping( @@ -454,6 +452,7 @@ def _normalize_file_value( config=upload_config, strict_type_validation=True, access_controller=_file_access_controller, + sessions=self._session_factory, ) except ValueError as exc: raise HumanInputSubmissionValidationError( @@ -461,9 +460,8 @@ def _normalize_file_value( ) from exc return cast(JsonValue, file.to_dict()) - @classmethod def _normalize_file_list_value( - cls, + self, *, tenant_id: str, form_input: FileListInputConfig, @@ -483,7 +481,7 @@ def _normalize_file_list_value( raise HumanInputSubmissionValidationError( f"Invalid value for file list input '{form_input.output_variable_name}': expected list of mappings" ) - upload_config = cls._build_file_upload_config( + upload_config = self._build_file_upload_config( form_input=form_input, number_limits=form_input.number_limits, ) @@ -495,6 +493,7 @@ def _normalize_file_list_value( config=upload_config, strict_type_validation=True, access_controller=_file_access_controller, + sessions=self._session_factory, ) except ValueError as exc: raise HumanInputSubmissionValidationError( @@ -519,10 +518,9 @@ def _build_file_upload_config( class ServiceAPIHumanInputService(HumanInputService): """Preserve legacy JSON paragraph values for app-token form submissions.""" - @classmethod @override def _normalize_input_value( - cls, + self, *, tenant_id: str, form_input: FormInputConfig, diff --git a/api/services/workflow_event_snapshot_service.py b/api/services/workflow_event_snapshot_service.py index ebfa92c448e985..a15310d0226347 100644 --- a/api/services/workflow_event_snapshot_service.py +++ b/api/services/workflow_event_snapshot_service.py @@ -13,8 +13,6 @@ from sqlalchemy import desc, select from sqlalchemy.orm import Session, sessionmaker -from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter -from core.app.apps.message_generator import MessageGenerator from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity from core.app.entities.task_entities import ( HumanInputRequiredResponse, @@ -55,6 +53,8 @@ from repositories.entities.workflow_pause import WorkflowPauseEntity from repositories.factory import DifyAPIRepositoryFactory from services.app_task_service import AppTaskService +from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter +from services.workflow.execution.adapters.response_stream import WorkflowEventStream from services.workflow_pause_service import WorkflowPauseSnapshot, load_workflow_pause_snapshot logger = logging.getLogger(__name__) @@ -97,7 +97,7 @@ def build_workflow_event_stream( close_on_pause: bool = True, include_node_details: bool = False, ) -> Generator[Mapping[str, Any] | str]: - topic = MessageGenerator.get_response_topic(app_mode, workflow_run.id) + topic = WorkflowEventStream.get_response_topic(app_mode, workflow_run.id) workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) node_execution_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(session_maker) diff --git a/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py b/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py index f39390762822ec..cb5de62b6608e8 100644 --- a/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py +++ b/api/tests/e2e_tests/containers/controllers/web/test_human_input_form.py @@ -30,7 +30,12 @@ HumanInputFormRecipient, StandaloneWebAppRecipientPayload, ) -from models.human_input_entities import FormDefinition, SelectInputConfig, StringListSource, UserActionConfig +from models.human_input_entities import ( + FormDefinition, + SelectInputConfig, + StringListSource, + UserActionConfig, +) from models.model import App, AppMode, CustomizeTokenStrategy, IconType, Site from models.workflow import WorkflowRun, WorkflowType from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository diff --git a/api/tests/unit_tests/controllers/openapi/test_human_input_form.py b/api/tests/unit_tests/controllers/openapi/test_human_input_form.py index 9807d648db2baa..216e4ed30b53b9 100644 --- a/api/tests/unit_tests/controllers/openapi/test_human_input_form.py +++ b/api/tests/unit_tests/controllers/openapi/test_human_input_form.py @@ -38,10 +38,10 @@ ) from controllers.openapi.workflow_events import OpenApiWorkflowEventsApi from core.app.entities.task_entities import HumanInputRequiredResponse -from core.workflow.nodes.human_input import ParagraphInputConfig, UserActionConfig from enums.human_input import RecipientType from models.account import Account from models.enums import CreatorUserRole, EndUserType +from models.human_input_entities import ParagraphInputConfig, UserActionConfig from models.model import App, AppMode, EndUser _MODULE = "controllers.openapi.human_input_form" diff --git a/api/tests/unit_tests/controllers/openapi/test_workflow_events.py b/api/tests/unit_tests/controllers/openapi/test_workflow_events.py index 36c4d7cfe578be..a226fc2611f2c9 100644 --- a/api/tests/unit_tests/controllers/openapi/test_workflow_events.py +++ b/api/tests/unit_tests/controllers/openapi/test_workflow_events.py @@ -16,8 +16,8 @@ import pytest from flask import Flask -from werkzeug.exceptions import NotFound +from controllers.common.errors import NotFoundError from controllers.openapi.workflow_events import OpenApiWorkflowEventsApi, WorkflowEventsQuery from core.db.session_factory import session_factory from graphon.enums import WorkflowExecutionStatus @@ -123,14 +123,14 @@ def _bind_generators(self, monkeypatch: pytest.MonkeyPatch) -> None: module = sys.modules["controllers.openapi.workflow_events"] msg_gen_mock = Mock() msg_gen_mock.retrieve_events.return_value = iter([]) - monkeypatch.setattr(module, "MessageGenerator", lambda: msg_gen_mock) + monkeypatch.setattr(module, "WorkflowEventStream", msg_gen_mock) def test_not_found_when_run_missing(self, app: Flask, monkeypatch: pytest.MonkeyPatch): factory_mock = self._bind_repo(monkeypatch, None) api = OpenApiWorkflowEventsApi() with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -149,7 +149,7 @@ def test_not_found_when_run_belongs_to_different_app(self, app: Flask, monkeypat api = OpenApiWorkflowEventsApi() with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -163,7 +163,7 @@ def test_account_caller_rejected_for_end_user_run(self, app: Flask, monkeypatch: api = OpenApiWorkflowEventsApi() with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -244,7 +244,7 @@ def test_stream_reads_everything_off_the_context_before_streaming( msg_gen_mock = Mock() msg_gen_mock.retrieve_events.return_value = iter(["a", "b"]) - monkeypatch.setattr(module, "MessageGenerator", lambda: msg_gen_mock) + monkeypatch.setattr(module, "WorkflowEventStream", msg_gen_mock) monkeypatch.setattr(module, "build_workflow_event_stream", lambda **_kwargs: iter(["a", "b"])) ctx = _context(_make_account(), CreatorUserRole.ACCOUNT) @@ -266,7 +266,7 @@ def test_continue_on_pause_comes_off_the_declared_query_model( self._bind_repo(monkeypatch, _make_workflow_run(created_by_role=CreatorUserRole.ACCOUNT, created_by="acct-1")) msg_gen_mock = Mock() msg_gen_mock.retrieve_events.return_value = iter([]) - monkeypatch.setattr(module, "MessageGenerator", lambda: msg_gen_mock) + monkeypatch.setattr(module, "WorkflowEventStream", msg_gen_mock) api = OpenApiWorkflowEventsApi() with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): diff --git a/api/tests/unit_tests/controllers/service_api/app/test_human_input_form.py b/api/tests/unit_tests/controllers/service_api/app/test_human_input_form.py index 98d5388b696332..d3bc8a98096a79 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_human_input_form.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_human_input_form.py @@ -12,8 +12,8 @@ import pytest from flask import Flask, request from sqlalchemy.engine import Engine -from werkzeug.exceptions import NotFound +from controllers.common.errors import NotFoundError from controllers.common.human_input import HumanInputFormSubmitPayload from controllers.service_api.app.human_input_form import WorkflowHumanInputFormApi from enums.human_input import RecipientType @@ -159,7 +159,7 @@ def test_get_form_not_in_app(self, app: Flask, monkeypatch: pytest.MonkeyPatch, app_model = _app() with app.test_request_context("/form/human_input/token-1", method="GET"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): handler(api, app_model=app_model, form_token="token-1") @pytest.mark.parametrize( @@ -193,7 +193,7 @@ def test_get_rejects_non_service_api_recipient_types( app_model = _app() with app.test_request_context("/form/human_input/token-1", method="GET"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): handler(api, app_model=app_model, form_token="token-1") service_mock.ensure_form_active.assert_not_called() @@ -333,7 +333,7 @@ def test_post_rejects_non_service_api_recipient_types( json={"inputs": {"name": "Alice"}, "action": "approve", "user": "external-1"}, ): payload = HumanInputFormSubmitPayload.model_validate(request.get_json() or {}) - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): handler(api, payload, app_model=app_model, end_user=end_user, form_token="token-1") service_mock.submit_form_by_token.assert_not_called() diff --git a/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py b/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py index 6fea76e5c35739..d8ef7f41d61497 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_workflow_events.py @@ -8,21 +8,18 @@ from inspect import unwrap from types import SimpleNamespace from typing import override -from unittest.mock import ANY, MagicMock, Mock +from unittest.mock import ANY, Mock import pytest from flask import Flask from sqlalchemy.engine import Engine from sqlalchemy.orm import Session, sessionmaker -from werkzeug.exceptions import NotFound import services.app_generate_service as ags_module +from controllers.common.errors import NotFoundError from controllers.service_api.app.error import NotWorkflowAppError from controllers.service_api.app.workflow_events import WorkflowEventsApi from core.app.app_config.entities import AppAdditionalFeatures, WorkflowUIBasedAppConfig -from core.app.apps.advanced_chat.generate_task_pipeline import ConversationSnapshot, MessageSnapshot, WorkflowSnapshot -from core.app.apps.common import workflow_response_converter -from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter from core.app.entities.app_invoke_entities import ( AdvancedChatAppGenerateEntity, InvokeFrom, @@ -41,21 +38,23 @@ from core.workflow.nodes.human_input.pause_reason import DifyHITLEventType, HumanInputRequired from core.workflow.system_variables import build_system_variables from enums import DeploymentEdition -from enums.human_input import FormInputType, HumanInputFormKind, HumanInputFormStatus +from enums.human_input import FormInputType, HumanInputFormKind, HumanInputFormStatus, RecipientType from graphon.entities import WorkflowStartReason from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus from graphon.runtime import RuntimeState, VariablePool from libs.datetime_utils import to_utc_timestamp from models.account import Account from models.enums import CreatorUserRole, EndUserType, MessageStatus, WorkflowRunTriggeredFrom -from models.human_input import HumanInputForm +from models.human_input import HumanInputForm, HumanInputFormRecipient from models.human_input_entities import FormDefinition, ParagraphInputConfig, UserActionConfig from models.model import App, AppMode, EndUser from models.workflow import Workflow, WorkflowRun, WorkflowType from models.workflow import WorkflowType as ModelWorkflowType from repositories.api_workflow_node_execution_repository import WorkflowNodeExecutionSnapshot from repositories.entities.workflow_pause import WorkflowPauseEntity +from services.app.generation.ports import ConversationSnapshot, MessageSnapshot, WorkflowSnapshot from services.app_generate_service import AppGenerateService +from services.workflow.execution.adapters.response_converter import WorkflowResponseConverter from services.workflow_event_snapshot_service import _build_paused_snapshot_events from tests.helpers.unit.config_override import apply_config_overrides @@ -163,7 +162,7 @@ def _persist_human_input_form( return form -def _build_service_api_pause_converter() -> WorkflowResponseConverter: +def _build_service_api_pause_converter(*, workflow_contexts, tool_providers) -> WorkflowResponseConverter: application_generate_entity = WorkflowAppGenerateEntity( user_from=UserFrom.END_USER, task_id="task", @@ -186,9 +185,11 @@ def _build_service_api_pause_converter() -> WorkflowResponseConverter: user = Account(name="Tester", email="tester@example.com") user.id = "account-id" return WorkflowResponseConverter( + contexts=workflow_contexts, application_generate_entity=application_generate_entity, user=user, system_variables=system_variables, + tool_providers=tool_providers, ) @@ -365,7 +366,7 @@ def test_workflow_events_continue_on_pause_keeps_stream_open( ) msg_generator = Mock() msg_generator.retrieve_events.return_value = [{"event": "streamed"}] - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) api = WorkflowEventsApi() handler = unwrap(api.get) @@ -402,7 +403,7 @@ def snapshot_builder(**kwargs: object) -> list[dict[str, str]]: snapshot_calls.append(kwargs) return [{"event": "snapshot"}] - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) monkeypatch.setattr(workflow_events_module, "build_workflow_event_stream", snapshot_builder) api = WorkflowEventsApi() @@ -446,10 +447,11 @@ def test_advanced_chat_blocking_injects_pause_state_config( sqlite_session_maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False) monkeypatch.setattr(ags_module.session_factory, "get_session_maker", lambda: sqlite_session_maker) - generator_instance = MagicMock() + generator_instance = Mock() generator_instance.generate.return_value = {"result": "advanced-blocking"} generator_instance.convert_to_event_stream.side_effect = lambda payload: payload - monkeypatch.setattr(ags_module, "AdvancedChatAppGenerator", lambda **_kwargs: generator_instance) + generator_factory = Mock(return_value=generator_instance) + monkeypatch.setattr(ags_module, "AdvancedChatAppGenerator", generator_factory) app_model = _hitl_service_api_app(app_id="app-id", tenant_id="tenant-id", mode=AppMode.ADVANCED_CHAT) user = _hitl_service_api_end_user(user_id="user-id", app_id="app-id", tenant_id="tenant-id") @@ -465,6 +467,7 @@ def test_advanced_chat_blocking_injects_pause_state_config( ) assert result == {"result": "advanced-blocking"} + generator_factory.assert_called_once_with(execution_driver=ANY) call_kwargs = generator_instance.generate.call_args.kwargs assert call_kwargs["streaming"] is False assert call_kwargs["pause_state_config"] is not None @@ -473,7 +476,9 @@ def test_advanced_chat_blocking_injects_pause_state_config( # Blocking payload contract def test_advanced_chat_blocking_pause_payload_contract(self) -> None: - from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter + from services.workflow.execution.adapters.chatflow.generate_response_converter import ( + AdvancedChatAppGenerateResponseConverter, + ) response = AdvancedChatAppGenerateResponseConverter.convert_blocking_full_response( _build_advanced_chat_paused_blocking_response() @@ -487,7 +492,9 @@ def test_advanced_chat_blocking_pause_payload_contract(self) -> None: assert "human_input_forms" not in response["data"] def test_workflow_blocking_pause_payload_contract(self) -> None: - from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter + from services.workflow.execution.adapters.workflow.generate_response_converter import ( + WorkflowAppGenerateResponseConverter, + ) response = WorkflowAppGenerateResponseConverter.convert_blocking_full_response( _build_workflow_paused_blocking_response() @@ -501,9 +508,13 @@ def test_workflow_blocking_pause_payload_contract(self) -> None: ] assert "human_input_forms" not in response["data"] - def test_advanced_chat_blocking_pipeline_pause_payload_contract(self) -> None: + def test_advanced_chat_blocking_pipeline_pause_payload_contract( + self, app_records, *, workflow_contexts, tool_providers + ) -> None: from core.app.app_config.entities import AppAdditionalFeatures - from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline + from services.workflow.execution.adapters.chatflow.generate_task_pipeline import ( + AdvancedChatAppGenerateTaskPipeline, + ) app_config = WorkflowUIBasedAppConfig( tenant_id="tenant", @@ -527,6 +538,8 @@ def test_advanced_chat_blocking_pipeline_pause_payload_contract(self) -> None: workflow_run_id="run-id", ) pipeline = AdvancedChatAppGenerateTaskPipeline( + contexts=workflow_contexts, + chat_records=app_records, application_generate_entity=application_generate_entity, workflow=WorkflowSnapshot(id="workflow-id", tenant_id="tenant", features_dict={}), queue_manager=SimpleNamespace(invoke_from=InvokeFrom.WEB_APP, graph_runtime_state=None), @@ -542,6 +555,7 @@ def test_advanced_chat_blocking_pipeline_pause_payload_contract(self) -> None: stream=False, dialogue_count=1, draft_var_saver_factory=lambda **kwargs: None, + tool_providers=tool_providers, ) pipeline._task_state.answer = "partial answer" pipeline._workflow_run_id = "run-id" @@ -594,8 +608,11 @@ def _gen(): assert response.data.reasons[0]["form_id"] == "form-1" assert response.data.reasons[0]["expiration_time"] == 123 - def test_workflow_blocking_pipeline_pause_payload_contract(self) -> None: - from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline + def test_workflow_blocking_pipeline_pause_payload_contract( + self, monkeypatch: pytest.MonkeyPatch, *, workflow_runtime, tool_providers + ) -> None: + from services.workflow.execution.adapters.workflow import generate_task_pipeline as workflow_pipeline_module + from services.workflow.execution.adapters.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline app_config = WorkflowUIBasedAppConfig( tenant_id="tenant", @@ -619,13 +636,17 @@ def test_workflow_blocking_pipeline_pause_payload_contract(self) -> None: call_depth=0, ) pipeline = WorkflowAppGenerateTaskPipeline( + contexts=workflow_runtime.contexts, + logs=workflow_runtime.logs, application_generate_entity=application_generate_entity, workflow=_workflow(workflow_id="workflow-id", app_id="app", tenant_id="tenant"), queue_manager=SimpleNamespace(invoke_from=InvokeFrom.WEB_APP, graph_runtime_state=None), user=_hitl_service_api_end_user(user_id="user", app_id="app", tenant_id="tenant"), stream=False, draft_var_saver_factory=lambda **kwargs: None, + tool_providers=tool_providers, ) + monkeypatch.setattr(workflow_pipeline_module.time, "time", lambda: 1700000000) def _gen(): yield HumanInputRequiredResponse( @@ -663,12 +684,12 @@ def _gen(): assert response.data.reasons == [{"TYPE": "human_input_required", "form_id": "form-1", "expiration_time": 1}] def test_service_api_pause_event_serializes_hitl_reason( - self, - monkeypatch: pytest.MonkeyPatch, - sqlite_engine: Engine, - sqlite_session: Session, + self, sqlite_session: Session, *, workflow_contexts, tool_providers ) -> None: - converter = _build_service_api_pause_converter() + converter = _build_service_api_pause_converter( + workflow_contexts=workflow_contexts, + tool_providers=tool_providers, + ) converter.workflow_start_to_stream_response( task_id="task", workflow_run_id="run-id", @@ -677,16 +698,20 @@ def test_service_api_pause_event_serializes_hitl_reason( ) expiration_time = datetime(2024, 1, 1, tzinfo=UTC) - _persist_human_input_form(sqlite_session, expiration_time=expiration_time) - - monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=sqlite_engine)) - monkeypatch.setattr( - workflow_response_converter, - "load_form_dispositions_by_form_id", - lambda form_ids, session=None, surface=None: { - "form-1": FormDisposition(form_token="token", approval_channels=[]) - }, + form = _persist_human_input_form(sqlite_session, expiration_time=expiration_time) + form.tenant_id = "tenant-id" + form.app_id = "app-id" + form.workflow_run_id = "run-id" + sqlite_session.add( + HumanInputFormRecipient( + form_id=form.id, + delivery_id="delivery-1", + recipient_type=RecipientType.STANDALONE_WEB_APP, + recipient_payload="{}", + access_token="token", + ) ) + sqlite_session.commit() reason = HumanInputRequired( form_id="form-1", @@ -902,7 +927,7 @@ def test_workflow_run_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatc end_user = _workflow_events_end_user() with app.test_request_context("/workflow/run-1/events?user=u1", method="GET"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): handler(api, app_model=app_model, end_user=end_user, workflow_run_id="run-1") def test_workflow_run_permission_denied( @@ -916,7 +941,7 @@ def test_workflow_run_permission_denied( end_user = _workflow_events_end_user() with app.test_request_context("/workflow/run-1/events?user=u1", method="GET"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): handler(api, app_model=app_model, end_user=end_user, workflow_run_id="run-1") def test_finished_run_returns_sse(self, app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None: @@ -958,7 +983,7 @@ def test_running_run_streams_events( ) msg_generator = Mock() msg_generator.retrieve_events.return_value = [{"event": "streamed"}] - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) api = WorkflowEventsApi() handler = unwrap(api.get) @@ -989,7 +1014,7 @@ def snapshot_builder(**kwargs: object) -> list[dict[str, str]]: snapshot_calls.append(kwargs) return [{"event": "snapshot"}] - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) monkeypatch.setattr(workflow_events_module, "build_workflow_event_stream", snapshot_builder) api = WorkflowEventsApi() diff --git a/api/tests/unit_tests/controllers/web/test_human_input_form.py b/api/tests/unit_tests/controllers/web/test_human_input_form.py index ece4584a97a233..40ab5805935be5 100644 --- a/api/tests/unit_tests/controllers/web/test_human_input_form.py +++ b/api/tests/unit_tests/controllers/web/test_human_input_form.py @@ -11,10 +11,9 @@ from flask import Flask from sqlalchemy import Engine from sqlalchemy.orm import Session -from werkzeug.exceptions import Forbidden import controllers.web.human_input_form as human_input_module -from controllers.common.errors import NotFoundError +from controllers.common.errors import AccessDeniedError, NotFoundError from controllers.web.error import WebFormRateLimitExceededError from core.repositories.human_input_repository import HumanInputFormRecord from enums.human_input import HumanInputFormKind, HumanInputFormStatus, RecipientType, ValueSourceType @@ -398,7 +397,7 @@ def get_definition(self): def test_get_form_raises_forbidden_when_site_missing( monkeypatch: pytest.MonkeyPatch, app: Flask, database_session: Session ): - """GET raises Forbidden if site cannot be resolved.""" + """GET raises AccessDeniedError if site cannot be resolved.""" expiration_time = datetime(2099, 1, 3, tzinfo=UTC) _, app_model, _ = _persist_app_site(database_session, include_site=False) @@ -435,7 +434,7 @@ def get_definition(self): monkeypatch.setattr(human_input_module, "HumanInputService", lambda engine: service_mock) with app.test_request_context("/api/form/human_input/token-1", method="GET"): - with pytest.raises(Forbidden): + with pytest.raises(AccessDeniedError): HumanInputFormApi().get("token-1") limiter_mock.is_rate_limited.assert_called_once_with("203.0.113.10") limiter_mock.increment_rate_limit.assert_called_once_with("203.0.113.10") @@ -444,7 +443,7 @@ def get_definition(self): def test_get_form_raises_forbidden_when_tenant_archived( monkeypatch: pytest.MonkeyPatch, app: Flask, database_session: Session ) -> None: - """GET raises Forbidden when the app's tenant is archived.""" + """GET raises AccessDeniedError when the app's tenant is archived.""" expiration_time = datetime(2099, 1, 3, tzinfo=UTC) tenant, app_model, _ = _persist_app_site(database_session) @@ -483,7 +482,7 @@ def get_definition(self) -> _FakeDefinition: monkeypatch.setattr(human_input_module, "HumanInputService", lambda engine: service_mock) with app.test_request_context("/api/form/human_input/token-1", method="GET"): - with pytest.raises(Forbidden): + with pytest.raises(AccessDeniedError): HumanInputFormApi().get("token-1") diff --git a/api/tests/unit_tests/services/test_human_input_service.py b/api/tests/unit_tests/services/test_human_input_service.py index 2174d6b25d29a0..ac32ceacc6e9d2 100644 --- a/api/tests/unit_tests/services/test_human_input_service.py +++ b/api/tests/unit_tests/services/test_human_input_service.py @@ -16,10 +16,6 @@ from core.app.app_config.entities import WorkflowUIBasedAppConfig from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext, _WorkflowGenerateEntityWrapper -from core.repositories.human_input_repository import ( - HumanInputFormRecord, - HumanInputFormSubmissionRepository, -) from enums.human_input import ( DeliveryMethodType, HumanInputFormKind, @@ -31,6 +27,7 @@ from graphon.runtime import RuntimeState, VariablePool from libs.datetime_utils import naive_utc_now from models.human_input import HumanInputDelivery, HumanInputForm, HumanInputFormRecipient +from models.human_input_contracts import HumanInputFormRecord from models.human_input_entities import ( FileInputConfig, FileListInputConfig, @@ -42,6 +39,7 @@ ) from models.model import App, AppMode from models.workflow import WorkflowRun +from repositories.human_input.form_repository import HumanInputFormSubmissionRepository from services.human_input_service import ( Form, FormExpiredError, @@ -105,13 +103,8 @@ def sample_form_record() -> HumanInputFormRecord: @pytest.fixture def form_repository( sqlite_session_factory: sessionmaker[Session], - monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, ) -> Callable[[HumanInputFormRecord | None], HumanInputFormSubmissionRepository]: - monkeypatch.setattr( - "core.repositories.human_input_repository.session_factory.create_session", sqlite_session_factory - ) - def build(record: HumanInputFormRecord | None) -> HumanInputFormSubmissionRepository: """Persist the form and recipient; None leaves the database empty for missing-token tests.""" if record is not None: @@ -154,7 +147,7 @@ def build(record: HumanInputFormRecord | None) -> HumanInputFormSubmissionReposi ) with sqlite_session_factory.begin() as session: session.add_all([form, delivery, recipient]) - repository = HumanInputFormSubmissionRepository() + repository = HumanInputFormSubmissionRepository(sessions=sqlite_session_factory) mocker.spy(repository, "get_by_token") mocker.spy(repository, "mark_submitted") return repository @@ -394,6 +387,7 @@ def test_submit_form_by_token_calls_repository_and_enqueue( persisted = repo.get_by_form_id(sample_form_record.form_id) assert persisted is not None assert persisted.status == HumanInputFormStatus.SUBMITTED + assert persisted.selected_action_id == "submit" assert persisted.submitted_data == {"field": "value"} assert persisted.submission_end_user_id == "end-user-id" enqueue_spy.assert_called_once_with(sample_form_record.workflow_run_id) diff --git a/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py b/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py index 2dc52dc56b664f..8a3e13b8c44e63 100644 --- a/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py +++ b/api/tests/unit_tests/services/test_workflow_event_snapshot_service.py @@ -221,7 +221,11 @@ def test_build_snapshot_events_applies_message_context() -> None: (None, None, "run-1"), ], ) -def test_resolve_task_id_priority(context_task_id, buffered_task_id, expected) -> None: +def test_resolve_task_id_priority( + context_task_id: str | None, + buffered_task_id: str | None, + expected: str, +) -> None: resumption_context = _build_resumption_context(context_task_id) if context_task_id else None buffer_state = BufferState( queue=queue.Queue(), @@ -812,7 +816,7 @@ def get_executions(tenant_id: str, app_id: str, workflow_run_id: str) -> Sequenc service_module.DifyAPIRepositoryFactory, "create_api_workflow_node_execution_repository", lambda _: node_repo ) monkeypatch.setattr( - service_module.MessageGenerator, "get_response_topic", lambda *_args: _Topic(_StaticSubscription()) + service_module.WorkflowEventStream, "get_response_topic", lambda *_args: _Topic(_StaticSubscription()) ) buffer_state = BufferState(queue.Queue(), Event(), Event(), Event(), task_id_hint="task-1") buffer_state.done_event.set() @@ -866,7 +870,7 @@ def test_build_workflow_event_stream_should_emit_ping_and_terminal_snapshot_even create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", MagicMock(return_value=topic)) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock(return_value=topic)) message_context_lookup = MagicMock(side_effect=lambda *_args, **_kwargs: call_order.append("message") or None) app_lookup = MagicMock(return_value=None) monkeypatch.setattr( @@ -961,6 +965,7 @@ def test_build_advanced_chat_snapshot_requires_conversation_context( create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock()) monkeypatch.setattr(service_module, "_load_resumption_context", MagicMock(return_value=resumption_context)) conversation_lookup = MagicMock(return_value=None) app_lookup = MagicMock(return_value=None) @@ -997,6 +1002,7 @@ def test_build_non_suspended_advanced_chat_snapshot_uses_app_scoped_fallback( create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock()) load_resumption_context = MagicMock(return_value=None) monkeypatch.setattr(service_module, "_load_resumption_context", load_resumption_context) conversation_lookup = MagicMock(return_value=None) @@ -1044,7 +1050,7 @@ def test_build_workflow_event_stream_should_emit_periodic_ping_and_stop_after_id create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", MagicMock(return_value=topic)) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock(return_value=topic)) monkeypatch.setattr(service_module, "_load_resumption_context", MagicMock(return_value=None)) monkeypatch.setattr(service_module, "_build_snapshot_events", MagicMock(return_value=[])) monkeypatch.setattr(service_module, "_resolve_task_id", MagicMock(return_value="task-1")) @@ -1099,7 +1105,7 @@ def test_build_workflow_event_stream_should_exit_when_buffer_done_and_empty( create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", MagicMock(return_value=topic)) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock(return_value=topic)) monkeypatch.setattr(service_module, "_load_resumption_context", MagicMock(return_value=None)) monkeypatch.setattr(service_module, "_build_snapshot_events", MagicMock(return_value=[])) monkeypatch.setattr(service_module, "_resolve_task_id", MagicMock(return_value="task-1")) @@ -1143,7 +1149,7 @@ def test_build_workflow_event_stream_should_continue_when_pause_loading_fails( create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", MagicMock(return_value=topic)) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock(return_value=topic)) monkeypatch.setattr(service_module, "_load_resumption_context", MagicMock(return_value=None)) monkeypatch.setattr(service_module, "_resolve_task_id", MagicMock(return_value="task-1")) snapshot_builder = MagicMock(return_value=[{"event": StreamEvent.WORKFLOW_FINISHED}]) @@ -1392,7 +1398,7 @@ def test_build_workflow_event_stream_loads_pause_tokens_without_flask_app_contex create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), ) monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", MagicMock(return_value=topic)) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", MagicMock(return_value=topic)) monkeypatch.setattr( service_module, "_load_resumption_context", MagicMock(return_value=_build_resumption_context("task-1")) ) @@ -1535,7 +1541,7 @@ def test_stream_continues_with_live_events_when_all_forms_submitted( monkeypatch.setattr(factory, "create_api_workflow_run_repository", lambda _sm: _PausedRunRepo(pause_entity)) monkeypatch.setattr(factory, "create_api_workflow_node_execution_repository", lambda _sm: _EmptyNodeRepo()) topic = _Topic(_StaticSubscription()) - monkeypatch.setattr(service_module.MessageGenerator, "get_response_topic", lambda *_args: topic) + monkeypatch.setattr(service_module.WorkflowEventStream, "get_response_topic", lambda *_args: topic) resumption_context = _build_resumption_context("task-1") monkeypatch.setattr(service_module, "_load_resumption_context", lambda _pause_entity: resumption_context) live_finished = {"event": StreamEvent.WORKFLOW_FINISHED.value, "task_id": "task-1"}