fix(api): fix workflow state persistence issue (#31752)

Ensure workflow pause configuration is correctly set for all entrypoints.

api/configs/feature/__init__.py

@@ -93,9 +93,9 @@ class AppExecutionConfig(BaseSettings):
         default=0,
     )
 
-    HITL_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
+    HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
         description="Maximum seconds a workflow run can stay paused waiting for human input before global timeout.",
-        default=int(timedelta(days=3).total_seconds()),
+        default=int(timedelta(days=7).total_seconds()),
         ge=1,
     )
 

api/core/plugin/backwards_invocation/app.py

@@ -12,6 +12,7 @@ from core.app.apps.chat.app_generator import ChatAppGenerator
 from core.app.apps.completion.app_generator import CompletionAppGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
 from extensions.ext_database import db
 from models import Account
@@ -102,6 +103,11 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             if not workflow:
                 raise ValueError("unexpected app type")
 
+            pause_config = PauseStateLayerConfig(
+                session_factory=db.engine,
+                state_owner_user_id=workflow.created_by,
+            )
+
             return AdvancedChatAppGenerator().generate(
                 app_model=app,
                 workflow=workflow,
@@ -115,6 +121,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                 invoke_from=InvokeFrom.SERVICE_API,
                 workflow_run_id=str(uuid.uuid4()),
                 streaming=stream,
+                pause_state_config=pause_config,
             )
         elif app.mode == AppMode.AGENT_CHAT:
             return AgentChatAppGenerator().generate(
@@ -161,6 +168,11 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         if not workflow:
             raise ValueError("unexpected app type")
 
+        pause_config = PauseStateLayerConfig(
+            session_factory=db.engine,
+            state_owner_user_id=workflow.created_by,
+        )
+
         return WorkflowAppGenerator().generate(
             app_model=app,
             workflow=workflow,
@@ -169,6 +181,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             invoke_from=InvokeFrom.SERVICE_API,
             streaming=stream,
             call_depth=1,
+            pause_state_config=pause_config,
         )
 
     @classmethod

api/core/tools/workflow_as_tool/tool.py

@@ -98,6 +98,10 @@ class WorkflowTool(Tool):
             invoke_from=self.runtime.invoke_from,
             streaming=False,
             call_depth=self.workflow_call_depth + 1,
+            # NOTE(QuantumGhost): We explicitly set `pause_state_config` to `None`
+            # because workflow pausing mechanisms (such as HumanInput) are not
+            # supported within WorkflowTool execution context.
+            pause_state_config=None,
         )
         assert isinstance(result, dict)
         data = result.get("data", {})

api/pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
     "numpy~=1.26.4",
     "openpyxl~=3.1.5",
     "opik~=1.8.72",
-    "litellm==1.77.1", # Pinned to avoid madoka dependency issue
+    "litellm==1.77.1",                                  # Pinned to avoid madoka dependency issue
     "opentelemetry-api==1.27.0",
     "opentelemetry-distro==0.48b0",
     "opentelemetry-exporter-otlp==1.27.0",
@@ -230,3 +230,23 @@ vdb = [
     "mo-vector~=0.1.13",
     "mysql-connector-python>=9.3.0",
 ]
+
+[tool.mypy]
+
+[[tool.mypy.overrides]]
+# targeted ignores for current type-check errors
+# TODO(QuantumGhost): suppress type errors in HITL related code.
+# fix the type error later
+module = [
+    "configs.middleware.cache.redis_pubsub_config",
+    "extensions.ext_redis",
+    "tasks.workflow_execution_tasks",
+    "core.workflow.nodes.base.node",
+    "services.human_input_delivery_test_service",
+    "core.app.apps.advanced_chat.app_generator",
+    "controllers.console.human_input_form",
+    "controllers.console.app.workflow_run",
+    "repositories.sqlalchemy_api_workflow_node_execution_repository",
+    "extensions.logstore.repositories.logstore_api_workflow_run_repository",
+]
+ignore_errors = true

api/services/app_generate_service.py

@@ -16,6 +16,8 @@ from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
 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 enums.quota_type import QuotaType, unlimited
 from extensions.otel import AppGenerateHandler, trace_span
 from models.model import Account, App, AppMode, EndUser
@@ -189,6 +191,10 @@ class AppGenerateService:
                         request_id,
                     )
 
+                pause_config = PauseStateLayerConfig(
+                    session_factory=session_factory.get_session_maker(),
+                    state_owner_user_id=workflow.created_by,
+                )
                 return rate_limit.generate(
                     WorkflowAppGenerator.convert_to_event_stream(
                         WorkflowAppGenerator().generate(
@@ -200,6 +206,7 @@ class AppGenerateService:
                             streaming=False,
                             root_node_id=root_node_id,
                             call_depth=0,
+                            pause_state_config=pause_config,
                         ),
                     ),
                     request_id,

api/services/human_input_service.py

@@ -239,7 +239,7 @@ class HumanInputService:
         logger.warning("App mode %s does not support resume for workflow run %s", app.mode, workflow_run_id)
 
     def _is_globally_expired(self, form: Form, *, now: datetime | None = None) -> bool:
-        global_timeout_seconds = dify_config.HITL_GLOBAL_TIMEOUT_SECONDS
+        global_timeout_seconds = dify_config.HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS
         if global_timeout_seconds <= 0:
             return False
         if form.workflow_run_id is None:

api/tasks/human_input_timeout_tasks.py

@@ -61,7 +61,7 @@ def check_and_handle_human_input_timeouts(limit: int = 100) -> None:
     form_repo = HumanInputFormSubmissionRepository(session_factory)
     service = HumanInputService(session_factory, form_repository=form_repo)
     now = naive_utc_now()
-    global_timeout_seconds = dify_config.HITL_GLOBAL_TIMEOUT_SECONDS
+    global_timeout_seconds = dify_config.HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS
 
     with session_factory() as session:
         global_deadline = now - timedelta(seconds=global_timeout_seconds) if global_timeout_seconds > 0 else None

api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py

@@ -0,0 +1,72 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig
+from core.plugin.backwards_invocation.app import PluginAppBackwardsInvocation
+from models.model import AppMode
+
+
+def test_invoke_chat_app_advanced_chat_injects_pause_state_config(mocker):
+    workflow = MagicMock()
+    workflow.created_by = "owner-id"
+
+    app = MagicMock()
+    app.mode = AppMode.ADVANCED_CHAT
+    app.workflow = workflow
+
+    mocker.patch(
+        "core.plugin.backwards_invocation.app.db",
+        SimpleNamespace(engine=MagicMock()),
+    )
+    generator_spy = mocker.patch(
+        "core.plugin.backwards_invocation.app.AdvancedChatAppGenerator.generate",
+        return_value={"result": "ok"},
+    )
+
+    result = PluginAppBackwardsInvocation.invoke_chat_app(
+        app=app,
+        user=MagicMock(),
+        conversation_id="conv-1",
+        query="hello",
+        stream=False,
+        inputs={"k": "v"},
+        files=[],
+    )
+
+    assert result == {"result": "ok"}
+    call_kwargs = generator_spy.call_args.kwargs
+    pause_state_config = call_kwargs.get("pause_state_config")
+    assert isinstance(pause_state_config, PauseStateLayerConfig)
+    assert pause_state_config.state_owner_user_id == "owner-id"
+
+
+def test_invoke_workflow_app_injects_pause_state_config(mocker):
+    workflow = MagicMock()
+    workflow.created_by = "owner-id"
+
+    app = MagicMock()
+    app.mode = AppMode.WORKFLOW
+    app.workflow = workflow
+
+    mocker.patch(
+        "core.plugin.backwards_invocation.app.db",
+        SimpleNamespace(engine=MagicMock()),
+    )
+    generator_spy = mocker.patch(
+        "core.plugin.backwards_invocation.app.WorkflowAppGenerator.generate",
+        return_value={"result": "ok"},
+    )
+
+    result = PluginAppBackwardsInvocation.invoke_workflow_app(
+        app=app,
+        user=MagicMock(),
+        stream=False,
+        inputs={"k": "v"},
+        files=[],
+    )
+
+    assert result == {"result": "ok"}
+    call_kwargs = generator_spy.call_args.kwargs
+    pause_state_config = call_kwargs.get("pause_state_config")
+    assert isinstance(pause_state_config, PauseStateLayerConfig)
+    assert pause_state_config.state_owner_user_id == "owner-id"

api/tests/unit_tests/core/tools/workflow_as_tool/test_tool.py

@@ -55,6 +55,43 @@ def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_fiel
     assert exc_info.value.args == ("oops",)
 
 
+def test_workflow_tool_does_not_use_pause_state_config(monkeypatch: pytest.MonkeyPatch):
+    entity = ToolEntity(
+        identity=ToolIdentity(author="test", name="test tool", label=I18nObject(en_US="test tool"), provider="test"),
+        parameters=[],
+        description=None,
+        has_runtime_parameters=False,
+    )
+    runtime = ToolRuntime(tenant_id="test_tool", invoke_from=InvokeFrom.EXPLORE)
+    tool = WorkflowTool(
+        workflow_app_id="",
+        workflow_as_tool_id="",
+        version="1",
+        workflow_entities={},
+        workflow_call_depth=1,
+        entity=entity,
+        runtime=runtime,
+    )
+
+    monkeypatch.setattr(tool, "_get_app", lambda *args, **kwargs: None)
+    monkeypatch.setattr(tool, "_get_workflow", lambda *args, **kwargs: None)
+
+    from unittest.mock import MagicMock, Mock
+
+    mock_user = Mock()
+    monkeypatch.setattr(tool, "_resolve_user", lambda *args, **kwargs: mock_user)
+
+    generate_mock = MagicMock(return_value={"data": {}})
+    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
+    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
+
+    list(tool.invoke("test_user", {}))
+
+    call_kwargs = generate_mock.call_args.kwargs
+    assert "pause_state_config" in call_kwargs
+    assert call_kwargs["pause_state_config"] is None
+
+
 def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch: pytest.MonkeyPatch):
     """Test that WorkflowTool should generate variable messages when there are outputs"""
     entity = ToolEntity(

api/tests/unit_tests/services/test_app_generate_service.py

@@ -0,0 +1,65 @@
+from unittest.mock import MagicMock
+
+import services.app_generate_service as app_generate_service_module
+from models.model import AppMode
+from services.app_generate_service import AppGenerateService
+
+
+class _DummyRateLimit:
+    def __init__(self, client_id: str, max_active_requests: int) -> None:
+        self.client_id = client_id
+        self.max_active_requests = max_active_requests
+
+    @staticmethod
+    def gen_request_key() -> str:
+        return "dummy-request-id"
+
+    def enter(self, request_id: str | None = None) -> str:
+        return request_id or "dummy-request-id"
+
+    def exit(self, request_id: str) -> None:
+        return None
+
+    def generate(self, generator, request_id: str):
+        return generator
+
+
+def test_workflow_blocking_injects_pause_state_config(mocker, monkeypatch):
+    monkeypatch.setattr(app_generate_service_module.dify_config, "BILLING_ENABLED", False)
+    mocker.patch("services.app_generate_service.RateLimit", _DummyRateLimit)
+
+    workflow = MagicMock()
+    workflow.id = "workflow-id"
+    workflow.created_by = "owner-id"
+
+    mocker.patch.object(AppGenerateService, "_get_workflow", return_value=workflow)
+
+    generator_spy = mocker.patch(
+        "services.app_generate_service.WorkflowAppGenerator.generate",
+        return_value={"result": "ok"},
+    )
+
+    app_model = MagicMock()
+    app_model.mode = AppMode.WORKFLOW
+    app_model.id = "app-id"
+    app_model.tenant_id = "tenant-id"
+    app_model.max_active_requests = 0
+    app_model.is_agent = False
+
+    user = MagicMock()
+    user.id = "user-id"
+
+    result = AppGenerateService.generate(
+        app_model=app_model,
+        user=user,
+        args={"inputs": {"k": "v"}},
+        invoke_from=MagicMock(),
+        streaming=False,
+    )
+
+    assert result == {"result": "ok"}
+
+    call_kwargs = generator_spy.call_args.kwargs
+    pause_state_config = call_kwargs.get("pause_state_config")
+    assert pause_state_config is not None
+    assert pause_state_config.state_owner_user_id == "owner-id"

api/tests/unit_tests/services/test_human_input_service.py

@@ -100,7 +100,7 @@ def test_ensure_form_active_respects_global_timeout(monkeypatch, sample_form_rec
         created_at=datetime.utcnow() - timedelta(hours=2),
         expiration_time=datetime.utcnow() + timedelta(hours=2),
     )
-    monkeypatch.setattr(human_input_service_module.dify_config, "HITL_GLOBAL_TIMEOUT_SECONDS", 3600)
+    monkeypatch.setattr(human_input_service_module.dify_config, "HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS", 3600)
 
     with pytest.raises(FormExpiredError):
         service.ensure_form_active(Form(expired_record))

api/tests/unit_tests/tasks/test_human_input_timeout_tasks.py

@@ -115,7 +115,7 @@ def test_is_global_timeout_uses_created_at():
 def test_check_and_handle_human_input_timeouts_marks_and_routes(monkeypatch: pytest.MonkeyPatch):
     now = datetime(2025, 1, 1, 12, 0, 0)
     monkeypatch.setattr(task_module, "naive_utc_now", lambda: now)
-    monkeypatch.setattr(task_module.dify_config, "HITL_GLOBAL_TIMEOUT_SECONDS", 3600)
+    monkeypatch.setattr(task_module.dify_config, "HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS", 3600)
     monkeypatch.setattr(task_module, "db", SimpleNamespace(engine=object()))
 
     forms = [
@@ -193,7 +193,7 @@ def test_check_and_handle_human_input_timeouts_marks_and_routes(monkeypatch: pyt
 def test_check_and_handle_human_input_timeouts_omits_global_filter_when_disabled(monkeypatch: pytest.MonkeyPatch):
     now = datetime(2025, 1, 1, 12, 0, 0)
     monkeypatch.setattr(task_module, "naive_utc_now", lambda: now)
-    monkeypatch.setattr(task_module.dify_config, "HITL_GLOBAL_TIMEOUT_SECONDS", 0)
+    monkeypatch.setattr(task_module.dify_config, "HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS", 0)
     monkeypatch.setattr(task_module, "db", SimpleNamespace(engine=object()))
 
     capture: dict[str, Any] = {}

api/ty.toml

@@ -43,4 +43,3 @@ exclude = [
     "controllers/web/workflow_events.py",
     "tasks/app_generate/workflow_execute_task.py",
 ]
-