From eaa5d39fd5d952c0430fa24a6bd36dfed4125850 Mon Sep 17 00:00:00 2001 From: Byron Wang Date: Thu, 8 Oct 2026 18:34:59 +0800 Subject: [PATCH 1/5] refactor(api): migrate Human Input forms and Workflow event streams --- api/controllers/console/human_input_form.py | 24 ++-- api/controllers/openapi/workflow_events.py | 30 ++--- .../service_api/app/human_input_form.py | 14 +-- .../service_api/app/workflow_events.py | 31 ++---- api/controllers/web/human_input_form.py | 11 +- api/controllers/web/workflow_events.py | 24 ++-- api/services/human_input_service.py | 41 ++++--- .../workflow_event_snapshot_service.py | 4 +- .../controllers/web/test_human_input_form.py | 7 +- .../console/test_human_input_form.py | 7 +- .../openapi/test_human_input_form.py | 2 +- .../openapi/test_workflow_events_openapi.py | 20 ++-- .../service_api/app/test_hitl_service_api.py | 103 ++++++++++++------ .../service_api/app/test_human_input_form.py | 8 +- .../service_api/app/test_workflow_events.py | 18 +-- .../controllers/web/test_human_input_form.py | 10 +- .../services/test_human_input_service.py | 8 +- .../test_workflow_event_snapshot_service.py | 14 +-- 18 files changed, 191 insertions(+), 185 deletions(-) diff --git a/api/controllers/console/human_input_form.py b/api/controllers/console/human_input_form.py index 4cbf29a89f1ea5..455fb4d42caa64 100644 --- a/api/controllers/console/human_input_form.py +++ b/api/controllers/console/human_input_form.py @@ -25,11 +25,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.login import login_required @@ -38,7 +33,10 @@ 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, HumanInputService +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__) @@ -204,22 +202,16 @@ def _generate_finished_events() -> Generator[str, None, None]: event_generator = _generate_finished_events else: - msg_generator = MessageGenerator() - generator: 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=AppMode(app.mode), workflow_run=workflow_run, @@ -230,7 +222,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 6e2be3d15d0725..89e5022f0ab7c8 100644 --- a/api/controllers/openapi/workflow_events.py +++ b/api/controllers/openapi/workflow_events.py @@ -14,9 +14,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 @@ -83,7 +81,7 @@ def get(self, ctx: Context, app_id: str, task_id: str, *, query: WorkflowEventsQ 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. @@ -95,15 +93,15 @@ def get(self, ctx: Context, app_id: str, task_id: str, *, query: WorkflowEventsQ ) 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 @@ -123,20 +121,14 @@ def _generate_finished_events() -> Generator[str, None, None]: event_generator = _generate_finished_events else: - msg_generator = MessageGenerator() - generator: BaseAppGenerator - if app_mode == AppMode.ADVANCED_CHAT: - generator = AdvancedChatAppGenerator() - else: - generator = 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 +139,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 007c9a67462cc2..e15f5944731cd9 100644 --- a/api/controllers/service_api/app/human_input_form.py +++ b/api/controllers/service_api/app/human_input_form.py @@ -13,8 +13,8 @@ from flask import Response from flask_restx import Resource 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 @@ -65,14 +65,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") @service_api_ns.route("/form/human_input/") @@ -115,7 +115,7 @@ def get(self, app_model: App, form_token: str): service = HumanInputService(db.engine) form = service.get_form_by_token(form_token) if form is None: - raise NotFound("Form not found") + raise NotFoundError("Form not found") _ensure_form_belongs_to_app(form, app_model) _ensure_form_is_allowed_for_service_api(form) @@ -169,7 +169,7 @@ def post(self, payload: HumanInputFormSubmitPayload, app_model: App, end_user: E service = HumanInputService(db.engine) form = service.get_form_by_token(form_token) if form is None: - raise NotFound("Form not found") + raise NotFoundError("Form not found") _ensure_form_belongs_to_app(form, app_model) _ensure_form_is_allowed_for_service_api(form) @@ -177,7 +177,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( @@ -188,6 +188,6 @@ def post(self, payload: HumanInputFormSubmitPayload, app_model: App, end_user: E submission_end_user_id=end_user.id, ) except FormNotFoundError: - raise NotFound("Form not found") + raise NotFoundError("Form not found") return {}, 200 diff --git a/api/controllers/service_api/app/workflow_events.py b/api/controllers/service_api/app/workflow_events.py index 2752b62d82cd10..ab27f682e8f36f 100644 --- a/api/controllers/service_api/app/workflow_events.py +++ b/api/controllers/service_api/app/workflow_events.py @@ -9,25 +9,23 @@ 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 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 @@ -105,16 +103,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 @@ -133,13 +131,8 @@ def _generate_finished_events() -> Generator[str, None, None]: event_generator = _generate_finished_events else: - msg_generator = MessageGenerator() - generator: 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" @@ -148,7 +141,7 @@ def _generate_finished_events() -> Generator[str, None, 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, @@ -159,7 +152,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 81be98806c2224..32cb41565abdf4 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 @@ -150,7 +149,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", @@ -184,7 +183,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) @@ -274,10 +273,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 2954b9a81de836..b911b2bcccb3a1 100644 --- a/api/controllers/web/workflow_events.py +++ b/api/controllers/web/workflow_events.py @@ -13,15 +13,13 @@ 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 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) @@ -75,22 +73,16 @@ def _generate_finished_events() -> Generator[str, None, None]: event_generator = _generate_finished_events else: app_mode = AppMode.value_of(app_model.mode) - msg_generator = MessageGenerator() - generator: 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, @@ -100,7 +92,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/services/human_input_service.py b/api/services/human_input_service.py index 2193ab25242912..05fe799b6163fd 100644 --- a/api/services/human_input_service.py +++ b/api/services/human_input_service.py @@ -10,10 +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 ( - HumanInputFormRecord, - 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 @@ -22,6 +18,7 @@ from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool from libs.datetime_utils import ensure_naive_utc, naive_utc_now from libs.exception import BaseHTTPException +from models.human_input_contracts import HumanInputFormRecord from models.human_input_entities import ( FileInputConfig, FileListInputConfig, @@ -31,9 +28,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 tasks.app_generate.workflow_execute_task import resume_app_execution _file_access_controller = DatabaseFileAccessController() @@ -329,9 +329,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, @@ -344,7 +343,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, @@ -357,9 +356,8 @@ def validate_and_normalize_submission( ) return normalized_form_data - @classmethod def normalize_submission_data( - cls, + self, *, tenant_id: str, form_definition: FormDefinitionProtocol, @@ -370,7 +368,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], @@ -378,24 +376,23 @@ def normalize_submission_data( return normalized_form_data - @classmethod def _normalize_input_value( - cls, + self, *, tenant_id: str, form_input: FormInputConfig, value: Any, ) -> JsonValue: 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, @@ -420,9 +417,8 @@ def _normalize_select_value( ) return value - @classmethod def _normalize_file_value( - cls, + self, *, tenant_id: str, form_input: FileInputConfig, @@ -432,7 +428,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( @@ -441,6 +437,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( @@ -448,9 +445,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, @@ -470,7 +466,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, ) @@ -482,6 +478,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( diff --git a/api/services/workflow_event_snapshot_service.py b/api/services/workflow_event_snapshot_service.py index 5fb86206ec9481..872beebdcca70d 100644 --- a/api/services/workflow_event_snapshot_service.py +++ b/api/services/workflow_event_snapshot_service.py @@ -12,7 +12,6 @@ from sqlalchemy import desc, select from sqlalchemy.orm import Session, sessionmaker -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, @@ -50,6 +49,7 @@ from repositories.api_workflow_node_execution_repository import WorkflowNodeExecutionSnapshot from repositories.entities.workflow_pause import WorkflowPauseEntity from repositories.factory import DifyAPIRepositoryFactory +from services.workflow.execution.adapters.response_stream import WorkflowEventStream logger = logging.getLogger(__name__) @@ -83,7 +83,7 @@ def build_workflow_event_stream( ping_interval: float = 10.0, close_on_pause: bool = True, ) -> Generator[Mapping[str, Any] | str, None, None]: - 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/test_containers_integration_tests/controllers/web/test_human_input_form.py b/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py index 93bc181de3cebf..c3dde9ab5482eb 100644 --- a/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py +++ b/api/tests/test_containers_integration_tests/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/console/test_human_input_form.py b/api/tests/unit_tests/controllers/console/test_human_input_form.py index 98a17921b034e6..f3575e20ec7df5 100644 --- a/api/tests/unit_tests/controllers/console/test_human_input_form.py +++ b/api/tests/unit_tests/controllers/console/test_human_input_form.py @@ -378,8 +378,7 @@ class _RepoStub: def get_workflow_run_by_id_and_tenant_id(self, **_kwargs): return workflow_run - workflow_generator = Mock() - workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"]) + response_stream = Mock(return_value=iter(["data: snapshot\n\n"])) snapshot_builder = Mock(return_value=["snapshot-events"]) monkeypatch.setattr( @@ -392,8 +391,8 @@ def get_workflow_run_by_id_and_tenant_id(self, **_kwargs): lambda *_args, **_kwargs: app_model, ) monkeypatch.setattr( - "controllers.console.human_input_form.WorkflowAppGenerator", - lambda: workflow_generator, + "controllers.console.human_input_form.convert_to_event_stream", + response_stream, ) monkeypatch.setattr( "controllers.console.human_input_form.build_workflow_event_stream", 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 919509af6879ab..e6fbe24c31035e 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 @@ -34,10 +34,10 @@ with_form_hints, ) 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_openapi.py b/api/tests/unit_tests/controllers/openapi/test_workflow_events_openapi.py index 9049b5ff4c93d1..d0f2891e00bd92 100644 --- a/api/tests/unit_tests/controllers/openapi/test_workflow_events_openapi.py +++ b/api/tests/unit_tests/controllers/openapi/test_workflow_events_openapi.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 @@ -121,17 +121,17 @@ def _bind_generators(self, monkeypatch: pytest.MonkeyPatch) -> None: module = sys.modules["controllers.openapi.workflow_events"] generator_mock = Mock() generator_mock.convert_to_event_stream.return_value = iter([]) - monkeypatch.setattr(module, "WorkflowAppGenerator", lambda: generator_mock) + monkeypatch.setattr(module, "convert_to_event_stream", generator_mock.convert_to_event_stream) 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("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -150,7 +150,7 @@ def test_not_found_when_run_belongs_to_different_app(self, app: Flask, monkeypat api = OpenApiWorkflowEventsApi() with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -164,7 +164,7 @@ def test_account_caller_rejected_for_end_user_run(self, app: Flask, monkeypatch: api = OpenApiWorkflowEventsApi() with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): - with pytest.raises(NotFound): + with pytest.raises(NotFoundError): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), @@ -245,10 +245,10 @@ def test_stream_reads_everything_off_the_context_before_streaming( generator_mock = Mock() generator_mock.convert_to_event_stream.return_value = iter(["event: a\n\n", "event: b\n\n"]) - monkeypatch.setattr(module, "WorkflowAppGenerator", lambda: generator_mock) + monkeypatch.setattr(module, "convert_to_event_stream", generator_mock.convert_to_event_stream) 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) monkeypatch.setattr(module, "build_workflow_event_stream", Mock(return_value=iter([]))) ctx = _context(_make_account(), CreatorUserRole.ACCOUNT) @@ -270,10 +270,10 @@ 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")) generator_mock = Mock() generator_mock.convert_to_event_stream.return_value = iter([]) - monkeypatch.setattr(module, "WorkflowAppGenerator", lambda: generator_mock) + monkeypatch.setattr(module, "convert_to_event_stream", generator_mock.convert_to_event_stream) 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("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): diff --git a/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py b/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py index f972226660bbbf..7caf79930b02cb 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py @@ -10,7 +10,7 @@ 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 @@ -20,9 +20,6 @@ import services.app_generate_service as ags_module 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, WorkflowAppGenerateEntity from core.app.entities.queue_entities import QueueWorkflowPausedEvent from core.app.entities.task_entities import ( @@ -36,20 +33,24 @@ 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 GraphRuntimeState, VariablePool from libs.datetime_utils import to_utc_timestamp from models.account import Account from models.enums import CreatorUserRole, EndUserType, MessageStatus -from models.human_input import HumanInputForm +from models.human_input import HumanInputForm, HumanInputFormRecipient from models.human_input_entities import ParagraphInputConfig, UserActionConfig from models.model import App, AppMode, EndUser from models.workflow import Workflow, WorkflowRun, WorkflowType 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.execution.ports import WorkflowRuntime +from services.workflow.variable_contracts import WorkflowExecutionVariables from services.workflow_event_snapshot_service import _build_snapshot_events from tests.unit_tests.config_override import apply_config_overrides @@ -149,7 +150,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 = SimpleNamespace( inputs={}, files=[], @@ -165,9 +166,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, ) @@ -341,8 +344,10 @@ def test_workflow_events_continue_on_pause_keeps_stream_open( msg_generator.retrieve_events.return_value = ["raw-event"] workflow_generator = Mock() workflow_generator.convert_to_event_stream.return_value = iter(["data: streamed\n\n"]) - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) - monkeypatch.setattr(workflow_events_module, "WorkflowAppGenerator", lambda: workflow_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) + monkeypatch.setattr( + workflow_events_module, "convert_to_event_stream", workflow_generator.convert_to_event_stream + ) api = WorkflowEventsApi() handler = unwrap(api.get) @@ -377,8 +382,10 @@ def test_workflow_events_snapshot_continue_on_pause_keeps_pause_open( workflow_generator = Mock() workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"]) snapshot_builder = Mock(return_value=["snapshot-events"]) - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) - monkeypatch.setattr(workflow_events_module, "WorkflowAppGenerator", lambda: workflow_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) + monkeypatch.setattr( + workflow_events_module, "convert_to_event_stream", workflow_generator.convert_to_event_stream + ) monkeypatch.setattr(workflow_events_module, "build_workflow_event_stream", snapshot_builder) api = WorkflowEventsApi() @@ -412,6 +419,9 @@ def test_advanced_chat_blocking_injects_pause_state_config( self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, + *, + workflow_variables: WorkflowExecutionVariables, + workflow_runtime: WorkflowRuntime, ) -> None: apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY) monkeypatch.setattr(ags_module, "RateLimit", _DummyRateLimit) @@ -421,10 +431,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: generator_instance) + generator_factory = Mock(return_value=generator_instance) + monkeypatch.setattr(ags_module, "AdvancedChatAppGenerator", generator_factory) app_model = _app(app_id="app-id", tenant_id="tenant-id", mode=AppMode.ADVANCED_CHAT) user = _end_user(user_id="user-id", app_id="app-id", tenant_id="tenant-id") @@ -437,9 +448,14 @@ def test_advanced_chat_blocking_injects_pause_state_config( args={"workflow_id": None, "query": "hi", "inputs": {}}, invoke_from=InvokeFrom.SERVICE_API, streaming=False, + variables=workflow_variables, + runtime=workflow_runtime, ) assert result == {"result": "advanced-blocking"} + dependencies = generator_factory.call_args.kwargs + assert callable(dependencies["draft_variable_loader"]) + assert callable(dependencies["draft_variable_saver"]) call_kwargs = generator_instance.generate.call_args.kwargs assert call_kwargs["streaming"] is False assert call_kwargs["pause_state_config"] is not None @@ -448,7 +464,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() @@ -462,7 +480,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() @@ -476,9 +496,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", @@ -502,6 +526,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), @@ -517,6 +543,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" @@ -569,9 +596,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, monkeypatch: pytest.MonkeyPatch) -> None: - from core.app.apps.workflow import generate_task_pipeline as workflow_pipeline_module - 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", @@ -595,12 +624,15 @@ def test_workflow_blocking_pipeline_pause_payload_contract(self, monkeypatch: py 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=_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) @@ -640,12 +672,11 @@ 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", @@ -654,16 +685,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", 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 0c5e5c07b9fedd..7839409af3f1b9 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 5419a34cb8b154..332453a529f9cc 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 @@ -12,8 +12,8 @@ import pytest from flask import Flask from sqlalchemy.engine import Engine -from werkzeug.exceptions import NotFound +from controllers.common.errors import NotFoundError from controllers.service_api.app.error import NotWorkflowAppError from controllers.service_api.app.workflow_events import WorkflowEventsApi from graphon.enums import WorkflowExecutionStatus @@ -101,7 +101,7 @@ def test_workflow_run_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatc end_user = _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( @@ -115,7 +115,7 @@ def test_workflow_run_permission_denied( end_user = _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: @@ -155,8 +155,10 @@ def test_running_run_streams_events( msg_generator.retrieve_events.return_value = ["raw-event"] workflow_generator = Mock() workflow_generator.convert_to_event_stream.return_value = iter(["data: streamed\n\n"]) - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) - monkeypatch.setattr(workflow_events_module, "WorkflowAppGenerator", lambda: workflow_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) + monkeypatch.setattr( + workflow_events_module, "convert_to_event_stream", workflow_generator.convert_to_event_stream + ) api = WorkflowEventsApi() handler = unwrap(api.get) @@ -183,8 +185,10 @@ def test_running_run_with_snapshot( workflow_generator = Mock() workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"]) snapshot_builder = Mock(return_value=["snapshot-events"]) - monkeypatch.setattr(workflow_events_module, "MessageGenerator", lambda: msg_generator) - monkeypatch.setattr(workflow_events_module, "WorkflowAppGenerator", lambda: workflow_generator) + monkeypatch.setattr(workflow_events_module, "WorkflowEventStream", msg_generator) + monkeypatch.setattr( + workflow_events_module, "convert_to_event_stream", workflow_generator.convert_to_event_stream + ) 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 73112836d485e4..e5c8be7505a5da 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,9 +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 AccessDeniedError from controllers.web.error import WebFormRateLimitExceededError from enums.human_input import RecipientType, ValueSourceType from models import Tenant @@ -398,7 +398,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) @@ -433,7 +433,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") @@ -442,7 +442,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) @@ -479,7 +479,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 f90374c6e1f410..9a4fb5c1034098 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, 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 GraphRuntimeState, 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, @@ -109,7 +107,7 @@ def form_repository( mocker: MockerFixture, ) -> Callable[[HumanInputFormRecord | None], HumanInputFormSubmissionRepository]: monkeypatch.setattr( - "core.repositories.human_input_repository.session_factory.create_session", sqlite_session_factory + "repositories.human_input.form_repository.session_factory.create_session", sqlite_session_factory ) def build(record: HumanInputFormRecord | None) -> HumanInputFormSubmissionRepository: diff --git a/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py b/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py index b1eaebe96017d5..a2784653262e16 100644 --- a/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py +++ b/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py @@ -720,7 +720,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( @@ -815,7 +815,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.MessageGenerator, "get_response_topic", MagicMock()) + 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) @@ -852,7 +852,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.MessageGenerator, "get_response_topic", MagicMock()) + 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) @@ -900,7 +900,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")) @@ -955,7 +955,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")) @@ -999,7 +999,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}]) @@ -1273,7 +1273,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")) ) From e7ebbbf190666f0c00d162dbb999899c7f188bbb Mon Sep 17 00:00:00 2001 From: Byron Wang Date: Fri, 9 Oct 2026 13:28:30 +0800 Subject: [PATCH 2/5] fix: align workflow event tests with active service boundaries --- api/openapi/markdown/web-openapi.md | 2 +- .../service_api/app/test_hitl_service_api.py | 11 +---------- .../controllers/web/test_workflow_events.py | 5 ++++- 3 files changed, 6 insertions(+), 12 deletions(-) diff --git a/api/openapi/markdown/web-openapi.md b/api/openapi/markdown/web-openapi.md index 08bc2afc8171fe..c2823bd8066992 100644 --- a/api/openapi/markdown/web-openapi.md +++ b/api/openapi/markdown/web-openapi.md @@ -361,7 +361,7 @@ GET /api/form/human_input/ | Code | Description | Schema | | ---- | ----------- | ------ | | 200 | Form retrieved successfully | **application/json**: [HumanInputFormDefinitionResponse](#humaninputformdefinitionresponse)
| -| 403 | Forbidden | | +| 403 | AccessDeniedError | | | 404 | Form not found | | | 412 | Form already submitted or expired | | | 429 | Too many requests | | diff --git a/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py b/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py index 7caf79930b02cb..66ac14e5facff5 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py @@ -49,8 +49,6 @@ 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.execution.ports import WorkflowRuntime -from services.workflow.variable_contracts import WorkflowExecutionVariables from services.workflow_event_snapshot_service import _build_snapshot_events from tests.unit_tests.config_override import apply_config_overrides @@ -419,9 +417,6 @@ def test_advanced_chat_blocking_injects_pause_state_config( self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, - *, - workflow_variables: WorkflowExecutionVariables, - workflow_runtime: WorkflowRuntime, ) -> None: apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY) monkeypatch.setattr(ags_module, "RateLimit", _DummyRateLimit) @@ -448,14 +443,10 @@ def test_advanced_chat_blocking_injects_pause_state_config( args={"workflow_id": None, "query": "hi", "inputs": {}}, invoke_from=InvokeFrom.SERVICE_API, streaming=False, - variables=workflow_variables, - runtime=workflow_runtime, ) assert result == {"result": "advanced-blocking"} - dependencies = generator_factory.call_args.kwargs - assert callable(dependencies["draft_variable_loader"]) - assert callable(dependencies["draft_variable_saver"]) + generator_factory.assert_called_once_with() call_kwargs = generator_instance.generate.call_args.kwargs assert call_kwargs["streaming"] is False assert call_kwargs["pause_state_config"] is not None diff --git a/api/tests/unit_tests/controllers/web/test_workflow_events.py b/api/tests/unit_tests/controllers/web/test_workflow_events.py index e6a4abf86dfaea..0907b7eb057273 100644 --- a/api/tests/unit_tests/controllers/web/test_workflow_events.py +++ b/api/tests/unit_tests/controllers/web/test_workflow_events.py @@ -148,7 +148,9 @@ def test_snapshot_stream_can_continue_across_pauses( workflow_generator = Mock() workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"]) snapshot_builder = Mock(return_value=["snapshot-events"]) - monkeypatch.setattr("controllers.web.workflow_events.WorkflowAppGenerator", lambda: workflow_generator) + monkeypatch.setattr( + "controllers.web.workflow_events.convert_to_event_stream", workflow_generator.convert_to_event_stream + ) monkeypatch.setattr("controllers.web.workflow_events.build_workflow_event_stream", snapshot_builder) with app.test_request_context("/workflow/run-1/events?include_state_snapshot=true&continue_on_pause=true"): @@ -163,3 +165,4 @@ def test_snapshot_stream_can_continue_across_pauses( session_maker=ANY, close_on_pause=False, ) + workflow_generator.convert_to_event_stream.assert_called_once_with(["snapshot-events"]) From e0979b14f7bfd4aa939cf424b2674e5392e93eca Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 05:31:22 +0000 Subject: [PATCH 3/5] [autofix.ci] apply automated fixes --- .../generated/api/console/apps/errors.gen.ts | 26 ------------------- .../generated/api/console/errors.gen.ts | 20 -------------- .../generated/api/console/rag/errors.gen.ts | 18 ------------- 3 files changed, 64 deletions(-) diff --git a/packages/contracts/generated/api/console/apps/errors.gen.ts b/packages/contracts/generated/api/console/apps/errors.gen.ts index 1ca81cfa7c000b..4cd4d5e228e674 100644 --- a/packages/contracts/generated/api/console/apps/errors.gen.ts +++ b/packages/contracts/generated/api/console/apps/errors.gen.ts @@ -605,8 +605,6 @@ export const get15Errors = { app_not_found: { status: 404, data: zAppNotFoundError }, } -export const get16Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const post13Errors = { no_file_uploaded: { status: 400, data: zNoFileUploadedError }, too_many_files: { status: 400, data: zTooManyFilesError }, @@ -919,17 +917,10 @@ export const get65Errors = { app_not_found: { status: 404, data: zAppNotFoundErr export const get66Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } -export const get67Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const post44Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const get68Errors = { - app_not_found: { status: 404, data: zAppNotFoundError }, draft_workflow_not_exist: { status: 404, data: zDraftWorkflowNotExistError }, } -export const post45Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const post46Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } export const post47Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } @@ -993,10 +984,6 @@ export const post56Errors = { app_not_found: { status: 404, data: zAppNotFoundEr export const post57Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } -export const delete11Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const get72Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const post58Errors = { app_not_found: { status: 404, data: zAppNotFoundError }, rate_limit_error: { status: 429, data: zRateLimitError }, @@ -1025,8 +1012,6 @@ export const get76Errors = { workflow_run_not_found: { status: 404, data: zWorkflowRunNotFoundError }, } -export const get77Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const post59Errors = { app_not_found: { status: 404, data: zAppNotFoundError }, rate_limit_error: { status: 429, data: zRateLimitError }, @@ -1037,18 +1022,7 @@ export const post60Errors = { rate_limit_error: { status: 429, data: zRateLimitError }, } -export const put6Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const delete12Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const get78Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const patch2Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - -export const delete13Errors = { app_not_found: { status: 404, data: zAppNotFoundError } } - export const get79Errors = { - app_not_found: { status: 404, data: zAppNotFoundError }, draft_workflow_not_exist: { status: 404, data: zDraftWorkflowNotExistError }, } diff --git a/packages/contracts/generated/api/console/errors.gen.ts b/packages/contracts/generated/api/console/errors.gen.ts index 34e0844610ffc4..e3098084a0a65e 100644 --- a/packages/contracts/generated/api/console/errors.gen.ts +++ b/packages/contracts/generated/api/console/errors.gen.ts @@ -1,11 +1,8 @@ // This file is auto-generated by scripts/add-orpc-error-maps.ts export { delete10Errors as 'DELETE /apps/{app_id}/workflow/comments/{comment_id}' } from './apps/errors.gen.ts' export { delete10Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}/research-tasks/{task_id}' } from './knowledge-fs/errors.gen.ts' -export { delete11Errors as 'DELETE /apps/{app_id}/workflows/draft/nodes/{node_id}/variables' } from './apps/errors.gen.ts' export { delete11Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}/sources/{source_id}' } from './knowledge-fs/errors.gen.ts' -export { delete12Errors as 'DELETE /apps/{app_id}/workflows/draft/variables/{variable_id}' } from './apps/errors.gen.ts' export { delete12Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}' } from './knowledge-fs/errors.gen.ts' -export { delete13Errors as 'DELETE /apps/{app_id}/workflows/draft/variables' } from './apps/errors.gen.ts' export { delete13Errors as 'DELETE /knowledge-fs/uploads/{upload_id}' } from './knowledge-fs/errors.gen.ts' export { delete14Errors as 'DELETE /apps/{app_id}/workflows/{workflow_id}' } from './apps/errors.gen.ts' export { delete16Errors as 'DELETE /workspaces/current/skills/{skill_id}' } from './workspaces/errors.gen.ts' @@ -14,15 +11,12 @@ export { delete2Errors as 'DELETE /apps/{app_id}/agent/config/skills/{name}' } f export { delete2Errors as 'DELETE /datasets/external-knowledge-api/{external_knowledge_api_id}' } from './datasets/errors.gen.ts' export { delete2Errors as 'DELETE /installed-apps/{installed_app_id}/saved-messages/{message_id}' } from './installed-apps/errors.gen.ts' export { delete2Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}/app-bindings/{caller_kind}/{app_id}' } from './knowledge-fs/errors.gen.ts' -export { delete2Errors as 'DELETE /rag/pipelines/{pipeline_id}/workflows/draft/nodes/{node_id}/variables' } from './rag/errors.gen.ts' export { delete3Errors as 'DELETE /agent/{agent_id}/config/files/{name}' } from './agent/errors.gen.ts' export { delete3Errors as 'DELETE /datasets/{dataset_id}/documents/{document_id}/segments/{segment_id}/child_chunks/{child_chunk_id}' } from './datasets/errors.gen.ts' export { delete3Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}/documents/bulk' } from './knowledge-fs/errors.gen.ts' -export { delete3Errors as 'DELETE /rag/pipelines/{pipeline_id}/workflows/draft/variables/{variable_id}' } from './rag/errors.gen.ts' export { delete4Errors as 'DELETE /agent/{agent_id}/config/skills/{name}' } from './agent/errors.gen.ts' export { delete4Errors as 'DELETE /apps/{app_id}/annotations' } from './apps/errors.gen.ts' export { delete4Errors as 'DELETE /knowledge-fs/spaces/{control_space_id}/documents/{document_id}' } from './knowledge-fs/errors.gen.ts' -export { delete4Errors as 'DELETE /rag/pipelines/{pipeline_id}/workflows/draft/variables' } from './rag/errors.gen.ts' export { delete4Errors as 'DELETE /workspaces/current/members/{member_id}' } from './workspaces/errors.gen.ts' export { delete5Errors as 'DELETE /agent/{agent_id}/trace-config' } from './agent/errors.gen.ts' export { delete5Errors as 'DELETE /apps/{app_id}/chat-conversations/{conversation_id}' } from './apps/errors.gen.ts' @@ -63,7 +57,6 @@ export { get13Errors as 'GET /trial-apps/{app_id}/messages/{message_id}/suggeste export { get14Errors as 'GET /agent/{agent_id}/config/skills/{name}/files/content' } from './agent/errors.gen.ts' export { get14Errors as 'GET /apps/{app_id}/agent/config/skills/{name}/inspect' } from './apps/errors.gen.ts' export { get14Errors as 'GET /knowledge-fs/spaces/{control_space_id}/documents/{document_id}/revisions' } from './knowledge-fs/errors.gen.ts' -export { get14Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/nodes/{node_id}/variables' } from './rag/errors.gen.ts' export { get14Errors as 'GET /snippets/{snippet_id}/workflows/draft/variables' } from './snippets/errors.gen.ts' export { get14Errors as 'GET /trial-apps/{app_id}/parameters' } from './trial-apps/errors.gen.ts' export { get15Errors as 'GET /agent/{agent_id}/config/skills/{name}/files/download' } from './agent/errors.gen.ts' @@ -73,19 +66,16 @@ export { get15Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/pre-pr export { get15Errors as 'GET /snippets/{snippet_id}/workflows/draft' } from './snippets/errors.gen.ts' export { get15Errors as 'GET /trial-apps/{app_id}/site' } from './trial-apps/errors.gen.ts' export { get16Errors as 'GET /agent/{agent_id}/config/skills/{name}/files/preview' } from './agent/errors.gen.ts' -export { get16Errors as 'GET /apps/{app_id}/agent/logs' } from './apps/errors.gen.ts' export { get16Errors as 'GET /datasets/{dataset_id}/documents/{document_id}/indexing-estimate' } from './datasets/errors.gen.ts' export { get16Errors as 'GET /knowledge-fs/spaces/{control_space_id}/documents' } from './knowledge-fs/errors.gen.ts' export { get16Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/processing/parameters' } from './rag/errors.gen.ts' export { get16Errors as 'GET /trial-apps/{app_id}/workflows' } from './trial-apps/errors.gen.ts' export { get17Errors as 'GET /agent/{agent_id}/config/skills/{name}/inspect' } from './agent/errors.gen.ts' export { get17Errors as 'GET /knowledge-fs/spaces/{control_space_id}/external-access' } from './knowledge-fs/errors.gen.ts' -export { get17Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/system-variables' } from './rag/errors.gen.ts' export { get17Errors as 'GET /trial-apps/{app_id}' } from './trial-apps/errors.gen.ts' export { get18Errors as 'GET /agent/{agent_id}/config/skills' } from './agent/errors.gen.ts' export { get18Errors as 'GET /datasets/{dataset_id}/documents/{document_id}/notion/sync' } from './datasets/errors.gen.ts' export { get18Errors as 'GET /knowledge-fs/spaces/{control_space_id}/golden-questions' } from './knowledge-fs/errors.gen.ts' -export { get18Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/variables/{variable_id}' } from './rag/errors.gen.ts' export { get19Errors as 'GET /agent/{agent_id}/log-sources' } from './agent/errors.gen.ts' export { get19Errors as 'GET /knowledge-fs/spaces/{control_space_id}/jobs/{job_id}' } from './knowledge-fs/errors.gen.ts' export { get19Errors as 'GET /rag/pipelines/{pipeline_id}/workflows/draft/variables' } from './rag/errors.gen.ts' @@ -228,7 +218,6 @@ export { get63Errors as 'GET /apps/{app_id}/workflow/statistics/daily-terminals' export { get64Errors as 'GET /apps/{app_id}/workflow/statistics/token-costs' } from './apps/errors.gen.ts' export { get65Errors as 'GET /apps/{app_id}/workflows/default-workflow-block-configs/{block_type}' } from './apps/errors.gen.ts' export { get66Errors as 'GET /apps/{app_id}/workflows/default-workflow-block-configs' } from './apps/errors.gen.ts' -export { get67Errors as 'GET /apps/{app_id}/workflows/draft/conversation-variables' } from './apps/errors.gen.ts' export { get68Errors as 'GET /apps/{app_id}/workflows/draft/environment-variables' } from './apps/errors.gen.ts' export { get69Errors as 'GET /apps/{app_id}/workflows/draft/nodes/{node_id}/agent-composer/candidates' } from './apps/errors.gen.ts' export { get6Errors as 'GET /agent/{agent_id}/chat-messages' } from './agent/errors.gen.ts' @@ -239,13 +228,10 @@ export { get6Errors as 'GET /rag/pipelines/{pipeline_id}/exports' } from './rag/ export { get6Errors as 'GET /trial-apps/{app_id}/agent/config/skills/{name}/files/content' } from './trial-apps/errors.gen.ts' export { get70Errors as 'GET /apps/{app_id}/workflows/draft/nodes/{node_id}/agent-composer' } from './apps/errors.gen.ts' export { get71Errors as 'GET /apps/{app_id}/workflows/draft/nodes/{node_id}/last-run' } from './apps/errors.gen.ts' -export { get72Errors as 'GET /apps/{app_id}/workflows/draft/nodes/{node_id}/variables' } from './apps/errors.gen.ts' export { get73Errors as 'GET /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/events' } from './apps/errors.gen.ts' export { get74Errors as 'GET /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/{node_id}/{output_name}/preview' } from './apps/errors.gen.ts' export { get75Errors as 'GET /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/{node_id}' } from './apps/errors.gen.ts' export { get76Errors as 'GET /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs' } from './apps/errors.gen.ts' -export { get77Errors as 'GET /apps/{app_id}/workflows/draft/system-variables' } from './apps/errors.gen.ts' -export { get78Errors as 'GET /apps/{app_id}/workflows/draft/variables/{variable_id}' } from './apps/errors.gen.ts' export { get79Errors as 'GET /apps/{app_id}/workflows/draft/variables' } from './apps/errors.gen.ts' export { get7Errors as 'GET /account/profile' } from './account/errors.gen.ts' export { get7Errors as 'GET /agent/{agent_id}/composer/candidates' } from './agent/errors.gen.ts' @@ -287,11 +273,9 @@ export { getErrors as 'GET /installed-apps/{installed_app_id}/conversations' } f export { getErrors as 'GET /remote-files/{url}' } from './remote-files/errors.gen.ts' export { getErrors as 'GET /trial-apps/{app_id}/agent-composer' } from './trial-apps/errors.gen.ts' export { getErrors as 'GET /website/crawl/status/{job_id}' } from './website/errors.gen.ts' -export { patch2Errors as 'PATCH /apps/{app_id}/workflows/draft/variables/{variable_id}' } from './apps/errors.gen.ts' export { patch2Errors as 'PATCH /datasets/{dataset_id}/documents/status/{action}/batch' } from './datasets/errors.gen.ts' export { patch2Errors as 'PATCH /installed-apps/{installed_app_id}/conversations/{c_id}/unpin' } from './installed-apps/errors.gen.ts' export { patch2Errors as 'PATCH /knowledge-fs/spaces/{control_space_id}/golden-questions/{question_id}' } from './knowledge-fs/errors.gen.ts' -export { patch2Errors as 'PATCH /rag/pipelines/{pipeline_id}/workflows/draft/variables/{variable_id}' } from './rag/errors.gen.ts' export { patch3Errors as 'PATCH /apps/{app_id}/workflows/{workflow_id}' } from './apps/errors.gen.ts' export { patch3Errors as 'PATCH /datasets/{dataset_id}/documents/{document_id}/processing/pause' } from './datasets/errors.gen.ts' export { patch3Errors as 'PATCH /installed-apps/{installed_app_id}' } from './installed-apps/errors.gen.ts' @@ -452,9 +436,7 @@ export { post42Errors as 'POST /apps/{app_id}/workflow/comments/{comment_id}/res export { post42Errors as 'POST /knowledge-fs/uploads' } from './knowledge-fs/errors.gen.ts' export { post43Errors as 'POST /apps/{app_id}/workflow/comments' } from './apps/errors.gen.ts' export { post43Errors as 'POST /workspaces/current/plugin/upload/github' } from './workspaces/errors.gen.ts' -export { post44Errors as 'POST /apps/{app_id}/workflows/draft/conversation-variables' } from './apps/errors.gen.ts' export { post44Errors as 'POST /workspaces/current/plugin/upload/pkg' } from './workspaces/errors.gen.ts' -export { post45Errors as 'POST /apps/{app_id}/workflows/draft/environment-variables' } from './apps/errors.gen.ts' export { post46Errors as 'POST /apps/{app_id}/workflows/draft/features' } from './apps/errors.gen.ts' export { post47Errors as 'POST /apps/{app_id}/workflows/draft/human-input/nodes/{node_id}/delivery-test' } from './apps/errors.gen.ts' export { post48Errors as 'POST /apps/{app_id}/workflows/draft/human-input/nodes/{node_id}/form/preview' } from './apps/errors.gen.ts' @@ -559,10 +541,8 @@ export { put4Errors as 'PUT /apps/{app_id}/workflow/comments/{comment_id}' } fro export { put4Errors as 'PUT /knowledge-fs/spaces/{control_space_id}/sources/{source_id}/sync-policy' } from './knowledge-fs/errors.gen.ts' export { put5Errors as 'PUT /apps/{app_id}/workflows/draft/nodes/{node_id}/agent-composer' } from './apps/errors.gen.ts' export { put5Errors as 'PUT /knowledge-fs/spaces/{control_space_id}/tags' } from './knowledge-fs/errors.gen.ts' -export { put6Errors as 'PUT /apps/{app_id}/workflows/draft/variables/{variable_id}/reset' } from './apps/errors.gen.ts' export { putErrors as 'PUT /agent/{agent_id}/build-draft' } from './agent/errors.gen.ts' export { putErrors as 'PUT /apps/{app_id}/network-access-group' } from './apps/errors.gen.ts' export { putErrors as 'PUT /knowledge-fs/spaces/{control_space_id}/app-bindings' } from './knowledge-fs/errors.gen.ts' -export { putErrors as 'PUT /rag/pipelines/{pipeline_id}/workflows/draft/variables/{variable_id}/reset' } from './rag/errors.gen.ts' export { putErrors as 'PUT /snippets/{snippet_id}/workflows/draft/nodes/{node_id}/agent-composer' } from './snippets/errors.gen.ts' export { putErrors as 'PUT /workspaces/current/agents/{agent_id}/skills' } from './workspaces/errors.gen.ts' diff --git a/packages/contracts/generated/api/console/rag/errors.gen.ts b/packages/contracts/generated/api/console/rag/errors.gen.ts index f87498e70afbc6..501d339d7f2b47 100644 --- a/packages/contracts/generated/api/console/rag/errors.gen.ts +++ b/packages/contracts/generated/api/console/rag/errors.gen.ts @@ -78,7 +78,6 @@ export const post10Errors = { pipeline_not_found: { status: 404, data: zPipeline export const get12Errors = { draft_workflow_not_exist: { status: 404, data: zDraftWorkflowNotExistError }, - pipeline_not_found: { status: 404, data: zPipelineNotFoundError }, } export const post11Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } @@ -89,10 +88,6 @@ export const get13Errors = { pipeline_not_found: { status: 404, data: zPipelineN export const post13Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } -export const delete2Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const get14Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - export const get15Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } export const get16Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } @@ -102,21 +97,8 @@ export const post14Errors = { rate_limit_error: { status: 429, data: zRateLimitError }, } -export const get17Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const putErrors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const delete3Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const get18Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const patch2Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - -export const delete4Errors = { pipeline_not_found: { status: 404, data: zPipelineNotFoundError } } - export const get19Errors = { draft_workflow_not_exist: { status: 404, data: zDraftWorkflowNotExistError }, - pipeline_not_found: { status: 404, data: zPipelineNotFoundError }, } export const get20Errors = { From b1e27a54ecfc3541db733c3ff399ae4ee3895a4b Mon Sep 17 00:00:00 2001 From: Byron Wang Date: Sat, 31 Oct 2026 21:57:07 +0800 Subject: [PATCH 4/5] Regenerate Human Input event-stream API documentation --- api/openapi/markdown/web-openapi.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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)
| From 711a0047c5dba9b771097f148def5c9c5ab3a2fa Mon Sep 17 00:00:00 2001 From: Byron Wang Date: Sat, 31 Oct 2026 22:18:18 +0800 Subject: [PATCH 5/5] Preserve instance-based normalization for Service API human input --- api/services/human_input_service.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/api/services/human_input_service.py b/api/services/human_input_service.py index 139cb60a5ef8be..7f34c4b776f80a 100644 --- a/api/services/human_input_service.py +++ b/api/services/human_input_service.py @@ -518,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,