fix(api): recover persisted workflow draft file cleanup requests

api/configs/feature/__init__.py

@@ -1515,6 +1515,10 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable recovery of persisted workflow draft upload cleanup requests",
         default=True,
     )
+    WORKFLOW_DRAFT_FILE_CLEANUP_INTERVAL: PositiveInt = Field(
+        description="Workflow draft upload cleanup recovery interval in minutes",
+        default=5,
+    )
     ENABLE_APPLICATION_TRACE_RECONCILE_TASK: bool = Field(
         description="Enable periodic application trace recovery and retention cleanup when Agent tracing is enabled",
         default=True,

api/controllers/common/errors.py

@@ -215,14 +215,19 @@ class RequestBodyTooLargeError(BaseHTTPException):
     code = 413
 
 
+def reraise_http_error(error: Exception) -> None:
+    """Preserve framework and application HTTP errors at a transport boundary."""
+    if isinstance(error, HTTPException):
+        raise error
+
+
 def raise_unexpected_error(error: Exception, *, message: str) -> NoReturn:
     """Keep HTTP failures intact and sanitize unclassified failures at the transport boundary.
 
     Framework request parsing and project HTTP errors may both reach a route's
     error translator. Neither should be converted to a generic server error.
     """
-    if isinstance(error, HTTPException):
-        raise error
+    reraise_http_error(error)
     logging.getLogger(__name__).exception(message)
     raise InternalServerError() from error
 

api/controllers/mcp/mcp.py

@@ -344,6 +344,15 @@ def _handle_mcp_request(
             client_name = f"{client_info.name}@{client_info.version}"
             end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, session)
 
+        generate = AppGenerateService.generate
+        if isinstance(mcp_request.root, mcp_types.CallToolRequest):
+            services = application_services()
+            generate = partial(
+                AppGenerateService.generate,
+                variables=services.workflow_variables,
+                runtime=services.workflow_runtime,
+            )
+
         return handle_mcp_request(
             session,
             app,
@@ -353,9 +362,5 @@ def _handle_mcp_request(
             end_user,
             request_id,
             protocol_version,
-            generate=partial(
-                AppGenerateService.generate,
-                variables=application_services().workflow_variables,
-                runtime=application_services().workflow_runtime,
-            ),
+            generate=generate,
         )

api/controllers/openapi/app_workflow.py

@@ -66,7 +66,6 @@
 from models.workflow import Workflow, WorkflowRun
 from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 from services.workflow.contracts import DraftWorkflowMissingError, WorkflowRecord
-from services.workflow.dsl_import import stored_secret_ids
 from services.workflow.graph_check import CONTAINER_NODE_TYPES, GRAPH_MODES
 from services.workflow.graph_diff import draft_token
 from services.workflow_run_service import WorkflowRunListArgs
@@ -374,7 +373,8 @@ def put(self, ctx: Context, app_id: str, env_id: str, *, body: EnvVariableSetPay
         if body.value == encrypter.full_mask_token() and body.value_type != EnvVariableValueType.SECRET:
             raise SecretMaskNotSecret()
         [mapping] = Workflow.normalize_environment_variable_mappings([{**body.model_dump(mode="json"), "id": env_id}])
-        if mapping["value"] == HIDDEN_VALUE and env_id not in stored_secret_ids(draft):
+        secret_ids = {variable.id for variable in draft.environment_variables if isinstance(variable, SecretVariable)}
+        if mapping["value"] == HIDDEN_VALUE and env_id not in secret_ids:
             raise SecretMaskUnknownId()
         try:
             variable = variable_factory.build_environment_variable_from_mapping(mapping)

api/controllers/trigger/webhook.py

@@ -4,7 +4,7 @@
 from flask import jsonify, request
 
 from controllers.common.access_response import webhook_not_found_response
-from controllers.common.errors import RequestBodyTooLargeError
+from controllers.common.errors import RequestBodyTooLargeError, reraise_http_error
 from controllers.trigger import bp
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
@@ -163,6 +163,7 @@ def handle_webhook_debug(webhook_id: str):
         return webhook_not_found_response()
     except WebhookBodyTooLargeError as error:
         raise RequestBodyTooLargeError(str(error)) from error
-    except Exception:
+    except Exception as error:
+        reraise_http_error(error)
         logger.exception("Webhook debug processing failed for %s", webhook_id)
         return jsonify({"error": "Internal server error", "message": "An internal error has occurred."}), 500

api/controllers/web/completion.py

@@ -3,9 +3,10 @@
 
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.common.errors import InternalServerError, InvalidRequestError, NotFoundError
+from controllers.common.errors import InvalidRequestError, NotFoundError
 from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console.app.wraps import with_session

api/extensions/application_services/workflow.py

@@ -304,9 +304,7 @@ def build_console_workflow_service(
         ),
         apps=apps,
         presence=WorkflowCollaborationRepository(redis=redis),
-        access=WorkflowAccessGateway(
-            session_factory=database_client, apps=EnterpriseConsoleAppAccess(members=rbac_members)
-        ),
+        access=WorkflowAccessGateway(apps=EnterpriseConsoleAppAccess(members=rbac_members), members=rbac_members),
     )
 
 

api/extensions/ext_celery.py

@@ -220,12 +220,6 @@ def __call__(self, *args: object, **kwargs: object) -> object:
             "task": "tasks.workflow_debug_reservation_tasks.recover_workflow_debug_reservations",
             "schedule": timedelta(minutes=1),
         }
-    if dify_config.ENABLE_WORKFLOW_DRAFT_FILE_CLEANUP_TASK:
-        imports.append("tasks.workflow_draft_var_tasks")
-        beat_schedule["workflow_draft_file_cleanup"] = {
-            "task": "tasks.workflow_draft_var_tasks.recover_draft_variable_file_cleanup_task",
-            "schedule": timedelta(minutes=5),
-        }
     if dify_config.KNOWLEDGE_FS_BACKGROUND_WORKER_ENABLED is True:
         from services.knowledge_fs.background_contract import OPERATION_QUEUES, OPERATION_TASK
 
@@ -288,6 +282,12 @@ def __call__(self, *args: object, **kwargs: object) -> object:
                 "schedule": collection_schedule,
                 "options": {"expires": interval},
             }
+    if dify_config.ENABLE_WORKFLOW_DRAFT_FILE_CLEANUP_TASK:
+        imports.append("tasks.workflow_draft_var_tasks")
+        beat_schedule["workflow_draft_file_cleanup"] = {
+            "task": "tasks.workflow_draft_var_tasks.recover_draft_variable_file_cleanup_task",
+            "schedule": timedelta(minutes=dify_config.WORKFLOW_DRAFT_FILE_CLEANUP_INTERVAL),
+        }
     if dify_config.ENABLE_CONVERSATION_CLEANUP_TASK:
         imports.append("tasks.delete_conversation_task")
         beat_schedule["conversation_cleanup_sweeper"] = {

api/services/workflow/access_gateway.py

@@ -1,23 +1,27 @@
 """Workflow collaboration visibility and asset delivery adapters."""
 
-from sqlalchemy.orm import Session, sessionmaker
-
 from configs import dify_config
 from graphon.file import helpers as file_helpers
 from machinery.context import RequestContext
 from services.app.access import resolve_app_access_filter
 from services.app.console_service import ConsoleAppAccess
+from services.rbac.members import MemberService
 
 
 class WorkflowAccessGateway:
-    def __init__(self, *, session_factory: sessionmaker[Session], apps: ConsoleAppAccess) -> None:
-        self._sessions = session_factory
+    def __init__(self, *, apps: ConsoleAppAccess, members: MemberService) -> None:
         self._apps = apps
+        self._members = members
 
     def accessible_app_ids(self, context: RequestContext, maintainers: dict[str, str | None]) -> set[str]:
         if not dify_config.RBAC_ENABLED:
             return set(maintainers)
-        access = resolve_app_access_filter(context.active_workspace_id, context.account_id)
+        permissions = self._members.permissions(context.active_workspace_id, context.account_id)
+        access = resolve_app_access_filter(
+            context.active_workspace_id,
+            context.account_id,
+            permissions=permissions,
+        )
         return {
             app_id
             for app_id, maintainer in maintainers.items()

api/tests/unit_tests/controllers/console/app/test_workflow.py

@@ -91,8 +91,9 @@ def test_sync_invalid_input_does_not_dispatch(
                 invoke(controller.DraftWorkflowApi, controller.DraftWorkflowApi.post)
             assert error.value.code == status
         else:
-            with pytest.raises(InvalidArgumentError, match="Invalid JSON data"):
+            with pytest.raises(InvalidArgumentError, match="Invalid JSON data") as error:
                 invoke(controller.DraftWorkflowApi, controller.DraftWorkflowApi.post)
+            assert error.value.code == status
     workflows.sync.assert_not_called()
 
 
@@ -284,13 +285,14 @@ def test_convert_delegates_permissions_result(workflows: Mock) -> None:
 
 
 def test_update_empty_payload_does_not_dispatch(workflows: Mock) -> None:
-    with pytest.raises(InvalidArgumentError, match="No valid fields to update"):
+    with pytest.raises(InvalidArgumentError, match="No valid fields to update") as error:
         invoke(
             controller.WorkflowByIdApi,
             controller.WorkflowByIdApi.patch,
             controller.WorkflowUpdatePayload(),
             workflow_id="version",
         )
+    assert error.value.code == 400
     workflows.update.assert_not_called()
 
 

api/tests/unit_tests/controllers/web/test_completion.py

@@ -9,8 +9,8 @@
 import pytest
 from flask import Flask
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import InternalServerError
 
-from controllers.common.errors import InternalServerError
 from controllers.web.completion import ChatApi, ChatStopApi, CompletionApi, CompletionStopApi
 from controllers.web.error import (
     AgentNotPublishedError,

api/tests/unit_tests/extensions/test_draft_variable_cleanup_schedule.py

@@ -0,0 +1,36 @@
+"""Recovery remains scheduled even when initial cleanup delivery fails."""
+
+from datetime import timedelta
+
+import pytest
+from celery import current_app
+
+from dify_app import DifyApp
+from extensions.ext_celery import init_app
+from tests.helpers.unit.config_override import apply_config_overrides
+
+
+@pytest.mark.parametrize("enabled", [True, False])
+def test_draft_cleanup_recovery_schedule(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
+    restore_default = current_app.set_default
+    restore_current = current_app.set_current
+    apply_config_overrides(
+        monkeypatch,
+        ENABLE_WORKFLOW_DRAFT_FILE_CLEANUP_TASK=enabled,
+        WORKFLOW_DRAFT_FILE_CLEANUP_INTERVAL=7,
+    )
+    application = init_app(DifyApp(__name__))
+    try:
+        schedules = application.conf.beat_schedule
+        if enabled:
+            assert schedules["workflow_draft_file_cleanup"] == {
+                "task": "tasks.workflow_draft_var_tasks.recover_draft_variable_file_cleanup_task",
+                "schedule": timedelta(minutes=7),
+            }
+            assert "tasks.workflow_draft_var_tasks" in application.conf.imports
+        else:
+            assert "workflow_draft_file_cleanup" not in schedules
+    finally:
+        application.close()
+        restore_default()
+        restore_current()

api/tests/unit_tests/extensions/test_ext_celery.py

@@ -34,7 +34,7 @@ def restore_celery_app_state() -> Iterator[None]:
     class TestCelerySSLConfiguration:
         """Test suite for Celery SSL configuration."""
 
-        def test_get_celery_broker_transport_options_includes_global_keyprefix_for_redis(self):
+        def test_get_celery_broker_transport_options_includes_global_keyprefix_for_redis(self) -> None:
             mock_config = MagicMock()
             mock_config.CELERY_USE_SENTINEL = False
             mock_config.REDIS_KEY_PREFIX = "enterprise-a"
@@ -46,7 +46,7 @@ def test_get_celery_broker_transport_options_includes_global_keyprefix_for_redis
 
             assert result["global_keyprefix"] == "enterprise-a:"
 
-        def test_get_celery_broker_transport_options_omits_global_keyprefix_when_prefix_empty(self):
+        def test_get_celery_broker_transport_options_omits_global_keyprefix_when_prefix_empty(self) -> None:
             mock_config = MagicMock()
             mock_config.CELERY_USE_SENTINEL = False
             mock_config.REDIS_KEY_PREFIX = "   "
@@ -58,7 +58,7 @@ def test_get_celery_broker_transport_options_omits_global_keyprefix_when_prefix_
 
             assert "global_keyprefix" not in result
 
-        def test_get_celery_broker_transport_options_keeps_sentinel_and_adds_global_keyprefix(self):
+        def test_get_celery_broker_transport_options_keeps_sentinel_and_adds_global_keyprefix(self) -> None:
             mock_config = MagicMock()
             mock_config.CELERY_USE_SENTINEL = True
             mock_config.CELERY_SENTINEL_MASTER_NAME = "mymaster"
@@ -75,7 +75,7 @@ def test_get_celery_broker_transport_options_keeps_sentinel_and_adds_global_keyp
             assert result["sentinel_kwargs"]["password"] == "secret"
             assert result["global_keyprefix"] == "enterprise-a:"
 
-        def test_get_celery_ssl_options_when_ssl_disabled(self):
+        def test_get_celery_ssl_options_when_ssl_disabled(self) -> None:
             """Test SSL options when BROKER_USE_SSL is False."""
             from configs import DifyConfig
 
@@ -87,7 +87,7 @@ def test_get_celery_ssl_options_when_ssl_disabled(self):
                 result = get_celery_ssl_options()
                 assert result is None
 
-        def test_get_celery_ssl_options_when_broker_not_redis(self):
+        def test_get_celery_ssl_options_when_broker_not_redis(self) -> None:
             """Test SSL options when broker is not Redis."""
             mock_config = MagicMock()
             mock_config.CELERY_BROKER_URL = "amqp://localhost:5672"
@@ -98,7 +98,7 @@ def test_get_celery_ssl_options_when_broker_not_redis(self):
                 result = get_celery_ssl_options()
                 assert result is None
 
-        def test_get_celery_ssl_options_with_cert_none(self):
+        def test_get_celery_ssl_options_with_cert_none(self) -> None:
             """Test SSL options with CERT_NONE requirement."""
             mock_config = MagicMock()
             mock_config.CELERY_BROKER_URL = "redis://localhost:6379/0"
@@ -117,7 +117,7 @@ def test_get_celery_ssl_options_with_cert_none(self):
                 assert result["ssl_certfile"] is None
                 assert result["ssl_keyfile"] is None
 
-        def test_get_celery_ssl_options_with_cert_required(self):
+        def test_get_celery_ssl_options_with_cert_required(self) -> None:
             """Test SSL options with CERT_REQUIRED and certificates."""
             mock_config = MagicMock()
             mock_config.CELERY_BROKER_URL = "rediss://localhost:6380/0"
@@ -136,7 +136,7 @@ def test_get_celery_ssl_options_with_cert_required(self):
                 assert result["ssl_certfile"] == "/path/to/client.crt"
                 assert result["ssl_keyfile"] == "/path/to/client.key"
 
-        def test_get_celery_ssl_options_with_cert_optional(self):
+        def test_get_celery_ssl_options_with_cert_optional(self) -> None:
             """Test SSL options with CERT_OPTIONAL requirement."""
             mock_config = MagicMock()
             mock_config.CELERY_BROKER_URL = "redis://localhost:6379/0"
@@ -153,7 +153,7 @@ def test_get_celery_ssl_options_with_cert_optional(self):
                 assert result["ssl_cert_reqs"] == ssl.CERT_OPTIONAL
                 assert result["ssl_ca_certs"] == "/path/to/ca.crt"
 
-        def test_get_celery_ssl_options_with_invalid_cert_reqs(self):
+        def test_get_celery_ssl_options_with_invalid_cert_reqs(self) -> None:
             """Test SSL options with invalid cert requirement defaults to CERT_NONE."""
             mock_config = MagicMock()
             mock_config.CELERY_BROKER_URL = "redis://localhost:6379/0"
@@ -169,7 +169,7 @@ def test_get_celery_ssl_options_with_invalid_cert_reqs(self):
                 assert result is not None
                 assert result["ssl_cert_reqs"] == ssl.CERT_NONE  # Should default to CERT_NONE
 
-        def test_celery_init_applies_ssl_to_broker_and_backend(self):
+        def test_celery_init_applies_ssl_to_broker_and_backend(self) -> None:
             """Test that SSL options are applied to both broker and backend when using Redis."""
             mock_config = MagicMock()
             mock_config.HUMAN_INPUT_TIMEOUT_TASK_INTERVAL = 1
@@ -188,6 +188,7 @@ def test_celery_init_applies_ssl_to_broker_and_backend(self):
             # Mock all the scheduler configs
             mock_config.CELERY_BEAT_SCHEDULER_TIME = 1
             mock_config.AGENT_SANDBOX_METERING_ENABLED = False
+            mock_config.WORKFLOW_DRAFT_FILE_CLEANUP_INTERVAL = 5
             mock_config.ENABLE_WORKFLOW_DRAFT_FILE_CLEANUP_TASK = False
             mock_config.ENABLE_CONVERSATION_CLEANUP_TASK = False
             mock_config.CONVERSATION_CLEANUP_TASK_INTERVAL = 5
@@ -231,7 +232,7 @@ def test_celery_init_applies_ssl_to_broker_and_backend(self):
                 assert celery_app.conf["redis_backend_use_ssl"] is not None
 
         @pytest.mark.parametrize("draft_cleanup_enabled", [True, False])
-        def test_celery_init_registers_required_agent_and_conversation_tasks(self, draft_cleanup_enabled: bool):
+        def test_celery_init_registers_required_agent_and_conversation_tasks(self, draft_cleanup_enabled: bool) -> None:
             mock_config = MagicMock()
             mock_config.BROKER_USE_SSL = False
             mock_config.REDIS_KEY_PREFIX = "enterprise-a"
@@ -243,10 +244,11 @@ def test_celery_init_registers_required_agent_and_conversation_tasks(self, draft
             mock_config.LOG_FORMAT = "%(message)s"
             mock_config.LOG_TZ = "UTC"
             mock_config.LOG_FILE = None
-            mock_config.CELERY_TASK_ANNOTATIONS = {}
+            mock_config.CELERY_TASK_ANNOTATIONS = dict[str, object]()
 
             mock_config.CELERY_BEAT_SCHEDULER_TIME = 1
             mock_config.AGENT_SANDBOX_METERING_ENABLED = False
+            mock_config.WORKFLOW_DRAFT_FILE_CLEANUP_INTERVAL = 5
             mock_config.ENABLE_WORKFLOW_DRAFT_FILE_CLEANUP_TASK = draft_cleanup_enabled
             mock_config.ENABLE_CONVERSATION_CLEANUP_TASK = True
             mock_config.CONVERSATION_CLEANUP_TASK_INTERVAL = 5