diff --git a/api/controllers/openapi/_errors.py b/api/controllers/openapi/_errors.py index 92884dfcd50..a53a379d948 100644 --- a/api/controllers/openapi/_errors.py +++ b/api/controllers/openapi/_errors.py @@ -67,6 +67,7 @@ class OpenApiErrorCode(StrEnum): MEMBER_LICENSE_EXCEEDED = "member_license_exceeded" HUMAN_INPUT_FORM_NOT_FOUND = "form_not_found" RECIPIENT_SURFACE_MISMATCH = "recipient_surface_mismatch" + TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE = "trigger_workflow_service_mode_unavailable" class ErrorDetail(BaseModel): diff --git a/api/controllers/openapi/app_run.py b/api/controllers/openapi/app_run.py index 772513ad417..631a750f0ee 100644 --- a/api/controllers/openapi/app_run.py +++ b/api/controllers/openapi/app_run.py @@ -35,6 +35,7 @@ from controllers.service_api.app.error import ( ProviderModelCurrentlyNotSupportError, ProviderNotInitializeError, ProviderQuotaExceededError, + TriggerWorkflowServiceModeUnavailableError, ) from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError from core.app.apps.base_app_queue_manager import AppQueueManager @@ -57,6 +58,9 @@ from services.errors.app import ( WorkflowIdFormatError, WorkflowNotFoundError, ) +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) from services.errors.llm import InvokeRateLimitError logger = logging.getLogger(__name__) @@ -70,6 +74,8 @@ def _translate_service_errors() -> Generator[None, None, None]: raise NotFound(str(ex)) except (IsDraftWorkflowError, WorkflowIdFormatError) as ex: raise BadRequest(str(ex)) + except TriggerWorkflowServiceModeUnavailableServiceError: + raise TriggerWorkflowServiceModeUnavailableError() except services.errors.conversation.ConversationNotExistsError: raise NotFound("Conversation Not Exists.") except services.errors.conversation.ConversationCompletedError: diff --git a/api/controllers/service_api/app/error.py b/api/controllers/service_api/app/error.py index e6f97e98249..60959746d49 100644 --- a/api/controllers/service_api/app/error.py +++ b/api/controllers/service_api/app/error.py @@ -1,4 +1,8 @@ from libs.exception import BaseHTTPException +from services.errors.app import ( + TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE, + TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE, +) class AppUnavailableError(BaseHTTPException): @@ -37,6 +41,12 @@ class WorkflowVersionExecutionNotAllowedError(BaseHTTPException): code = 403 +class TriggerWorkflowServiceModeUnavailableError(BaseHTTPException): + error_code = TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE + description = TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE + code = 403 + + class ConversationCompletedError(BaseHTTPException): error_code = "conversation_completed" description = "The conversation has ended. Please start a new conversation." diff --git a/api/controllers/service_api/app/workflow.py b/api/controllers/service_api/app/workflow.py index 846a500adc3..841b030d9d5 100644 --- a/api/controllers/service_api/app/workflow.py +++ b/api/controllers/service_api/app/workflow.py @@ -28,6 +28,7 @@ from controllers.service_api.app.error import ( ProviderModelCurrentlyNotSupportError, ProviderNotInitializeError, ProviderQuotaExceededError, + TriggerWorkflowServiceModeUnavailableError, WorkflowVersionExecutionNotAllowedError, ) from controllers.service_api.schema import ( @@ -60,7 +61,14 @@ from models.model import App, AppMode, EndUser from repositories.factory import DifyAPIRepositoryFactory from services.app_generate_service import AppGenerateService from services.billing_service import BillingService -from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError +from services.errors.app import ( + IsDraftWorkflowError, + WorkflowIdFormatError, + WorkflowNotFoundError, +) +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) from services.errors.llm import InvokeRateLimitError from services.workflow_app_service import WorkflowAppService @@ -295,6 +303,10 @@ class WorkflowRunApi(Resource): "- `completion_request_error` : Workflow execution request failed.\n" "- `invalid_param` : Invalid parameter value." ), + 403: ( + "`trigger_workflow_service_mode_unavailable` : Trigger-entry workflows cannot be invoked through " + "Web App, Service API, OpenAPI, or MCP." + ), 429: ( "- `too_many_requests` : Too many concurrent requests for this app.\n" "- `rate_limit_error` : The upstream model provider rate limit was exceeded." @@ -355,6 +367,8 @@ class WorkflowRunApi(Resource): # response-contract:ignore compact_generate_response return helper.compact_generate_response(response) + except TriggerWorkflowServiceModeUnavailableServiceError: + raise TriggerWorkflowServiceModeUnavailableError() except ProviderTokenNotInitError as ex: raise ProviderNotInitializeError(ex.description) except QuotaExceededError: @@ -401,8 +415,10 @@ class WorkflowRunByIdApi(Resource): "- `invalid_param` : Required parameter missing or invalid." ), 403: ( - "`workflow_version_execution_not_allowed` : Workflow version execution is unavailable on the " - "current plan. Upgrade to a paid plan." + "- `workflow_version_execution_not_allowed` : Workflow version execution is unavailable on the " + "current plan. Upgrade to a paid plan.\n" + "- `trigger_workflow_service_mode_unavailable` : The selected workflow version uses a trigger entry " + "and cannot be invoked through Web App, Service API, OpenAPI, or MCP." ), 404: "`not_found` : Workflow not found.", 429: ( @@ -482,6 +498,8 @@ class WorkflowRunByIdApi(Resource): # response-contract:ignore compact_generate_response return helper.compact_generate_response(response) + except TriggerWorkflowServiceModeUnavailableServiceError: + raise TriggerWorkflowServiceModeUnavailableError() except WorkflowNotFoundError as ex: raise NotFound(str(ex)) except IsDraftWorkflowError as ex: diff --git a/api/controllers/web/error.py b/api/controllers/web/error.py index b0ab2f0334c..16253f06eee 100644 --- a/api/controllers/web/error.py +++ b/api/controllers/web/error.py @@ -1,4 +1,8 @@ from libs.exception import BaseHTTPException +from services.errors.app import ( + TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE, + TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE, +) class AppUnavailableError(BaseHTTPException): @@ -31,6 +35,12 @@ class NotWorkflowAppError(BaseHTTPException): code = 400 +class TriggerWorkflowServiceModeUnavailableError(BaseHTTPException): + error_code = TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE + description = TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE + code = 403 + + class ConversationCompletedError(BaseHTTPException): error_code = "conversation_completed" description = "The conversation has ended. Please start a new conversation." diff --git a/api/controllers/web/workflow.py b/api/controllers/web/workflow.py index 1e6d6e24d92..9dc728e2efd 100644 --- a/api/controllers/web/workflow.py +++ b/api/controllers/web/workflow.py @@ -14,6 +14,7 @@ from controllers.web.error import ( ProviderModelCurrentlyNotSupportError, ProviderNotInitializeError, ProviderQuotaExceededError, + TriggerWorkflowServiceModeUnavailableError, ) from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError from controllers.web.wraps import WebApiResource @@ -30,6 +31,9 @@ from graphon.model_runtime.errors.invoke import InvokeError from libs import helper from models.model import App, AppMode, EndUser from services.app_generate_service import AppGenerateService +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) from services.errors.llm import InvokeRateLimitError logger = logging.getLogger(__name__) @@ -78,6 +82,8 @@ class WorkflowRunApi(WebApiResource): # response-contract:ignore compact_generate_response return helper.compact_generate_response(response) + except TriggerWorkflowServiceModeUnavailableServiceError: + raise TriggerWorkflowServiceModeUnavailableError() except ProviderTokenNotInitError as ex: raise ProviderNotInitializeError(ex.description) except QuotaExceededError: diff --git a/api/core/mcp/server/streamable_http.py b/api/core/mcp/server/streamable_http.py index 7fd03788c7e..3b96c5da43e 100644 --- a/api/core/mcp/server/streamable_http.py +++ b/api/core/mcp/server/streamable_http.py @@ -12,6 +12,7 @@ from core.mcp import types as mcp_types from graphon.variables.input_entities import VariableEntity, VariableEntityType from models.model import App, AppMCPServer, AppMode, EndUser from services.app_generate_service import AppGenerateService +from services.errors.app import TriggerWorkflowServiceModeUnavailableError logger = logging.getLogger(__name__) @@ -93,11 +94,16 @@ def handle_mcp_request( result=result_data.model_dump(by_alias=True, mode="json", exclude_none=True), ) - def create_error_response(code: int, message: str) -> mcp_types.JSONRPCError: + def create_error_response( + code: int, + message: str, + *, + data: Mapping[str, Any] | None = None, + ) -> mcp_types.JSONRPCError: """Create error response with error code and message""" from core.mcp.types import ErrorData - error_data = ErrorData(code=code, message=message) + error_data = ErrorData(code=code, message=message, data=data) return mcp_types.JSONRPCError( jsonrpc="2.0", id=request_id, @@ -131,6 +137,12 @@ def handle_mcp_request( case _: return create_error_response(mcp_types.METHOD_NOT_FOUND, f"Method not found: {request_type.__name__}") + except TriggerWorkflowServiceModeUnavailableError as e: + return create_error_response( + mcp_types.INVALID_REQUEST, + str(e), + data={"code": e.error_code}, + ) except ValueError as e: logger.exception("Invalid params") return create_error_response(mcp_types.INVALID_PARAMS, str(e)) diff --git a/api/services/app_generate_service.py b/api/services/app_generate_service.py index 4b271c6e94e..a22a7ba5d47 100644 --- a/api/services/app_generate_service.py +++ b/api/services/app_generate_service.py @@ -21,11 +21,17 @@ from core.app.features.rate_limiting import RateLimit from core.app.features.rate_limiting.rate_limit import rate_limit_context from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig from core.db import session_factory +from core.trigger.constants import is_trigger_node_type from enums import DeploymentEdition, QuotaType from extensions.otel import AppGenerateHandler, trace_span from models.model import Account, App, AppMode, EndUser from models.workflow import Workflow, WorkflowRun -from services.errors.app import QuotaExceededError, WorkflowIdFormatError, WorkflowNotFoundError +from services.errors.app import ( + QuotaExceededError, + TriggerWorkflowServiceModeUnavailableError, + WorkflowIdFormatError, + WorkflowNotFoundError, +) from services.errors.llm import InvokeRateLimitError from services.quota_service import QuotaService, unlimited from services.workflow_service import WorkflowService @@ -34,6 +40,13 @@ from tasks.app_generate.workflow_execute_task import AppExecutionParams, workflo logger = logging.getLogger(__name__) SSE_TASK_START_FALLBACK_MS = 200 +_MANUAL_WORKFLOW_INVOKE_SOURCES = frozenset( + { + InvokeFrom.OPENAPI, + InvokeFrom.SERVICE_API, + InvokeFrom.WEB_APP, + } +) if TYPE_CHECKING: from controllers.console.app.workflow import LoopNodeRunPayload @@ -290,6 +303,7 @@ class AppGenerateService: case AppMode.WORKFLOW: workflow_id = args.get("workflow_id") workflow = cls._get_workflow(app_model, invoke_from, workflow_id, session=session) + cls._ensure_workflow_service_mode_available(workflow=workflow, invoke_from=invoke_from) if streaming: with rate_limit_context(rate_limit, request_id): payload = AppExecutionParams.new( @@ -343,6 +357,16 @@ class AppGenerateService: case _: raise ValueError(f"Invalid app mode {app_model.mode}") + @staticmethod + def _ensure_workflow_service_mode_available(*, workflow: Workflow, invoke_from: InvokeFrom) -> None: + if invoke_from not in _MANUAL_WORKFLOW_INVOKE_SOURCES: + return + + for _, node_data in workflow.walk_nodes(): + node_type = node_data.get("type") + if isinstance(node_type, str) and is_trigger_node_type(node_type): + raise TriggerWorkflowServiceModeUnavailableError() + @staticmethod def _get_max_active_requests(app: App) -> int: """ diff --git a/api/services/errors/app.py b/api/services/errors/app.py index c9e9df97dea..74c29b0857d 100644 --- a/api/services/errors/app.py +++ b/api/services/errors/app.py @@ -18,6 +18,21 @@ class WorkflowIdFormatError(Exception): pass +TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE = "trigger_workflow_service_mode_unavailable" +TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE = ( + "This workflow uses a trigger entry and cannot be invoked through Web App, Service API, OpenAPI, or MCP." +) + + +class TriggerWorkflowServiceModeUnavailableError(Exception): + """Raised when a trigger-entry Workflow is invoked through a manual service surface.""" + + error_code = TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_CODE + + def __init__(self) -> None: + super().__init__(TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE_MESSAGE) + + class QuotaExceededError(ValueError): """Raised when billing quota is exceeded for a feature.""" diff --git a/api/tests/unit_tests/controllers/openapi/test_app_run_rate_limit.py b/api/tests/unit_tests/controllers/openapi/test_app_run_rate_limit.py index d9c468d0a4e..d2787a57591 100644 --- a/api/tests/unit_tests/controllers/openapi/test_app_run_rate_limit.py +++ b/api/tests/unit_tests/controllers/openapi/test_app_run_rate_limit.py @@ -8,8 +8,12 @@ import pytest from werkzeug.exceptions import TooManyRequests from controllers.openapi.app_run import _translate_service_errors +from controllers.service_api.app.error import TriggerWorkflowServiceModeUnavailableError from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError from core.errors.error import AppInvokeQuotaExceededError +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) from services.errors.llm import InvokeRateLimitError @@ -27,3 +31,11 @@ def test_translate_maps_workflow_quota_to_rate_limit_error(): raise InvokeRateLimitError("workflow quota exhausted") assert exc.value.error_code == "rate_limit_error" assert exc.value.code == 429 + + +def test_translate_maps_trigger_workflow_to_stable_unavailable_error(): + with pytest.raises(TriggerWorkflowServiceModeUnavailableError) as exc: + with _translate_service_errors(): + raise TriggerWorkflowServiceModeUnavailableServiceError() + assert exc.value.error_code == "trigger_workflow_service_mode_unavailable" + assert exc.value.code == 403 diff --git a/api/tests/unit_tests/controllers/service_api/app/test_workflow.py b/api/tests/unit_tests/controllers/service_api/app/test_workflow.py index c1cf3539477..4216df5f4bd 100644 --- a/api/tests/unit_tests/controllers/service_api/app/test_workflow.py +++ b/api/tests/unit_tests/controllers/service_api/app/test_workflow.py @@ -27,7 +27,11 @@ from sqlalchemy.engine import Engine from sqlalchemy.orm import Session, sessionmaker from werkzeug.exceptions import BadRequest, NotFound -from controllers.service_api.app.error import NotWorkflowAppError, WorkflowVersionExecutionNotAllowedError +from controllers.service_api.app.error import ( + NotWorkflowAppError, + TriggerWorkflowServiceModeUnavailableError, + WorkflowVersionExecutionNotAllowedError, +) from controllers.service_api.app.workflow import ( AppQueueManager, GraphEngineManager, @@ -50,7 +54,13 @@ from models.model import App, AppMode, EndUser from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun, WorkflowType from services.app_generate_service import AppGenerateService from services.billing_service import BillingService -from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError +from services.errors.app import ( + IsDraftWorkflowError, + WorkflowNotFoundError, +) +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) from services.errors.llm import InvokeRateLimitError from services.workflow_app_service import WorkflowAppService @@ -581,6 +591,32 @@ class TestWorkflowRunApi: with pytest.raises(InvokeRateLimitHttpError): handler(api, session=sqlite_session, app_model=app_model, end_user=end_user) + def test_trigger_workflow_returns_stable_unavailable_error( + self, + app: Flask, + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + ) -> None: + monkeypatch.setattr( + AppGenerateService, + "generate", + Mock(side_effect=TriggerWorkflowServiceModeUnavailableServiceError()), + ) + api = WorkflowRunApi() + handler = unwrap(api.post) + + with app.test_request_context("/workflows/run", method="POST", json={"inputs": {}}): + with pytest.raises(TriggerWorkflowServiceModeUnavailableError) as exc_info: + handler( + api, + session=sqlite_session, + app_model=_make_app_model(), + end_user=_make_end_user(), + ) + + assert exc_info.value.code == 403 + assert exc_info.value.error_code == "trigger_workflow_service_mode_unavailable" + def test_sandbox_billing_does_not_gate_default_workflow_run( self, app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session ) -> None: @@ -609,6 +645,33 @@ class TestWorkflowRunApi: class TestWorkflowRunByIdApi: + def test_trigger_workflow_version_returns_stable_unavailable_error( + self, + app: Flask, + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + ) -> None: + monkeypatch.setattr( + AppGenerateService, + "generate", + Mock(side_effect=TriggerWorkflowServiceModeUnavailableServiceError()), + ) + api = WorkflowRunByIdApi() + handler = unwrap(api.post) + + with app.test_request_context("/workflows/w1/run", method="POST", json={"inputs": {}}): + with pytest.raises(TriggerWorkflowServiceModeUnavailableError) as exc_info: + handler( + api, + session=sqlite_session, + app_model=_make_app_model(), + end_user=_make_end_user(), + workflow_id=str(uuid.uuid4()), + ) + + assert exc_info.value.code == 403 + assert exc_info.value.error_code == "trigger_workflow_service_mode_unavailable" + def test_rejects_sandbox_plan_with_upgrade_error( self, app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session ) -> None: diff --git a/api/tests/unit_tests/controllers/web/test_workflow.py b/api/tests/unit_tests/controllers/web/test_workflow.py index 2013b1e8db5..711976ed441 100644 --- a/api/tests/unit_tests/controllers/web/test_workflow.py +++ b/api/tests/unit_tests/controllers/web/test_workflow.py @@ -11,11 +11,15 @@ from controllers.web.error import ( NotWorkflowAppError, ProviderNotInitializeError, ProviderQuotaExceededError, + TriggerWorkflowServiceModeUnavailableError, ) from controllers.web.workflow import WorkflowRunApi, WorkflowTaskStopApi from core.errors.error import ProviderTokenNotInitError, QuotaExceededError from models.enums import EndUserType from models.model import App, AppMode, EndUser +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, +) def _workflow_app() -> App: @@ -68,6 +72,26 @@ class TestWorkflowRunApi: with pytest.raises(ProviderNotInitializeError): WorkflowRunApi().post(_workflow_app(), _end_user()) + @patch( + "controllers.web.workflow.AppGenerateService.generate", + side_effect=TriggerWorkflowServiceModeUnavailableServiceError(), + ) + @patch("controllers.web.workflow.web_ns") + def test_trigger_workflow_returns_stable_unavailable_error( + self, + mock_ns: MagicMock, + mock_gen: MagicMock, + app: Flask, + ) -> None: + mock_ns.payload = {"inputs": {}} + + with app.test_request_context("/workflows/run", method="POST"): + with pytest.raises(TriggerWorkflowServiceModeUnavailableError) as exc_info: + WorkflowRunApi().post(_workflow_app(), _end_user()) + + assert exc_info.value.code == 403 + assert exc_info.value.error_code == "trigger_workflow_service_mode_unavailable" + @patch( "controllers.web.workflow.AppGenerateService.generate", side_effect=QuotaExceededError(), diff --git a/api/tests/unit_tests/core/mcp/server/test_streamable_http.py b/api/tests/unit_tests/core/mcp/server/test_streamable_http.py index 42a9df3538a..1878fe08413 100644 --- a/api/tests/unit_tests/core/mcp/server/test_streamable_http.py +++ b/api/tests/unit_tests/core/mcp/server/test_streamable_http.py @@ -22,6 +22,7 @@ from core.mcp.server.streamable_http import ( ) from graphon.variables.input_entities import VariableEntity, VariableEntityType from models.model import App, AppMCPServer, AppMode, EndUser +from services.errors.app import TriggerWorkflowServiceModeUnavailableError class TestHandleMCPRequest: @@ -157,6 +158,29 @@ class TestHandleMCPRequest: # Verify AppGenerateService was called mock_app_generate.generate.assert_called_once() + @patch("core.mcp.server.streamable_http.AppGenerateService") + def test_handle_call_tool_returns_trigger_workflow_business_error(self, mock_app_generate): + mock_call_request = Mock(spec=types.CallToolRequest) + mock_call_request.params = Mock() + mock_call_request.params.arguments = {"query": "test question"} + mock_call_request.id = 123 + self.mock_request.root = mock_call_request + mock_app_generate.generate.side_effect = TriggerWorkflowServiceModeUnavailableError() + + result = handle_mcp_request( + Mock(), + self.app, + self.mock_request, + self.user_input_form, + self.mcp_server, + self.end_user, + 123, + ) + + assert isinstance(result, types.JSONRPCError) + assert result.error.code == types.INVALID_REQUEST + assert result.error.data == {"code": "trigger_workflow_service_mode_unavailable"} + @patch("core.mcp.server.streamable_http.AppGenerateService") def test_handle_call_tool_request_threads_protocol_version(self, mock_app_generate): """The negotiated version reaches handle_call_tool through the dispatcher.""" diff --git a/api/tests/unit_tests/services/test_app_generate_service.py b/api/tests/unit_tests/services/test_app_generate_service.py index db80881b187..9f5d75ce877 100644 --- a/api/tests/unit_tests/services/test_app_generate_service.py +++ b/api/tests/unit_tests/services/test_app_generate_service.py @@ -27,7 +27,11 @@ from core.app.entities.app_invoke_entities import InvokeFrom from enums import DeploymentEdition, QuotaType from models.model import AppMode from services.app_generate_service import AppGenerateService -from services.errors.app import WorkflowIdFormatError, WorkflowNotFoundError +from services.errors.app import ( + TriggerWorkflowServiceModeUnavailableError, + WorkflowIdFormatError, + WorkflowNotFoundError, +) # --------------------------------------------------------------------------- @@ -79,10 +83,18 @@ def _make_user() -> MagicMock: return user -def _make_workflow(*, workflow_id: str = "workflow-id", created_by: str = "owner-id") -> MagicMock: +def _make_workflow( + *, + workflow_id: str = "workflow-id", + created_by: str = "owner-id", + node_types: tuple[str, ...] = (), +) -> MagicMock: workflow = MagicMock() workflow.id = workflow_id workflow.created_by = created_by + workflow.walk_nodes.return_value = [ + (f"node-{index}", {"type": node_type}) for index, node_type in enumerate(node_types) + ] return workflow @@ -462,6 +474,84 @@ class TestGenerate: assert call_kwargs.get("pause_state_config") is not None assert call_kwargs["pause_state_config"].state_owner_user_id == "owner-id" + @pytest.mark.parametrize( + "invoke_from", + [InvokeFrom.OPENAPI, InvokeFrom.SERVICE_API, InvokeFrom.WEB_APP], + ) + @pytest.mark.parametrize("node_type", ["trigger-plugin", "trigger-schedule", "trigger-webhook"]) + def test_trigger_workflow_rejects_manual_service_surfaces( + self, + invoke_from: InvokeFrom, + node_type: str, + mocker: MockerFixture, + ) -> None: + workflow = _make_workflow(node_types=(node_type,)) + mocker.patch.object(AppGenerateService, "_get_workflow", return_value=workflow) + generate = mocker.patch("services.app_generate_service.WorkflowAppGenerator.generate") + + with pytest.raises(TriggerWorkflowServiceModeUnavailableError): + AppGenerateService.generate( + app_model=_make_app(AppMode.WORKFLOW), + user=_make_user(), + args={"inputs": {}}, + invoke_from=invoke_from, + streaming=False, + session=MagicMock(), + ) + + generate.assert_not_called() + + def test_trigger_workflow_allows_trigger_execution(self, mocker: MockerFixture) -> None: + workflow = _make_workflow(node_types=("trigger-webhook",)) + mocker.patch.object(AppGenerateService, "_get_workflow", return_value=workflow) + generate = mocker.patch( + "services.app_generate_service.WorkflowAppGenerator.generate", + return_value={"result": "trigger"}, + ) + mocker.patch( + "services.app_generate_service.WorkflowAppGenerator.convert_to_event_stream", + side_effect=lambda value: value, + ) + + result = AppGenerateService.generate( + app_model=_make_app(AppMode.WORKFLOW), + user=_make_user(), + args={"inputs": {}}, + invoke_from=InvokeFrom.TRIGGER, + streaming=False, + session=MagicMock(), + ) + + assert result == {"result": "trigger"} + generate.assert_called_once() + + def test_specific_start_workflow_version_remains_runnable(self, mocker: MockerFixture) -> None: + workflow_id = str(uuid.uuid4()) + workflow = _make_workflow(workflow_id=workflow_id, node_types=("start",)) + get_workflow = mocker.patch.object(AppGenerateService, "_get_workflow", return_value=workflow) + mocker.patch( + "services.app_generate_service.WorkflowAppGenerator.generate", + return_value={"result": "version"}, + ) + mocker.patch( + "services.app_generate_service.WorkflowAppGenerator.convert_to_event_stream", + side_effect=lambda value: value, + ) + app = _make_app(AppMode.WORKFLOW) + session = MagicMock() + + result = AppGenerateService.generate( + app_model=app, + user=_make_user(), + args={"inputs": {}, "workflow_id": workflow_id}, + invoke_from=InvokeFrom.SERVICE_API, + streaming=False, + session=session, + ) + + assert result == {"result": "version"} + get_workflow.assert_called_once_with(app, InvokeFrom.SERVICE_API, workflow_id, session=session) + # -- WORKFLOW streaming ------------------------------------------------- def test_workflow_streaming(self, mocker: MockerFixture): workflow = _make_workflow() diff --git a/packages/contracts/generated/api/openapi/types.gen.ts b/packages/contracts/generated/api/openapi/types.gen.ts index d0a00673acf..ac9af4af710 100644 --- a/packages/contracts/generated/api/openapi/types.gen.ts +++ b/packages/contracts/generated/api/openapi/types.gen.ts @@ -355,6 +355,7 @@ export type OpenApiErrorCode = | 'request_entity_too_large' | 'too_many_files' | 'too_many_requests' + | 'trigger_workflow_service_mode_unavailable' | 'unauthorized' | 'unknown' | 'unsupported_file_type' diff --git a/packages/contracts/generated/api/openapi/zod.gen.ts b/packages/contracts/generated/api/openapi/zod.gen.ts index fb584d54e9d..2cdc210b903 100644 --- a/packages/contracts/generated/api/openapi/zod.gen.ts +++ b/packages/contracts/generated/api/openapi/zod.gen.ts @@ -446,6 +446,7 @@ export const zOpenApiErrorCode = z.enum([ 'request_entity_too_large', 'too_many_files', 'too_many_requests', + 'trigger_workflow_service_mode_unavailable', 'unauthorized', 'unknown', 'unsupported_file_type',