fix(api): honor workflow node lifecycle contracts

api/core/workflow/llm_node.py

@@ -59,6 +59,12 @@ class DifyParameterExtractorNode(ParameterExtractorNode):
 
     _moderation_error_type: str = ""
 
+    @classmethod
+    @override
+    def version(cls) -> str:
+        """Return the explicit registry version required for every Dify node implementation."""
+        return "1"
+
     @override
     def _run(self) -> NodeRunResult:
         """Run normal extraction, preserving a caught moderation type without changing its verdict or outputs."""

api/tests/unit_tests/core/workflow/nodes/human_input_v2/test_node.py

@@ -6,7 +6,10 @@ from unittest.mock import MagicMock, create_autospec, patch
 import pytest
 from pydantic import SecretStr, TypeAdapter
 
+from core.app.apps.execution_coordinator import AppExecutionState
+from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
+from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     QueueHumanInputFormFilledEvent,
     QueueHumanInputFormTimeoutEvent,
@@ -273,11 +276,15 @@ def test_reentry_consumes_the_new_persisted_outcome(form: Form) -> None:
 def test_actual_graph_queue_response_path_preserves_order_and_fields(form: Form, status: HumanInputFormStatus) -> None:
     form = submit(form) if status == HumanInputFormStatus.SUBMITTED else replace(form, status=status)
     node, _ = build_node(form)
-    runner = create_autospec(WorkflowBasedAppRunner, instance=True)
+    queue = WorkflowAppQueueManager(
+        task_id="task-1", user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, app_mode="workflow"
+    )
+    runner = WorkflowBasedAppRunner(queue_manager=queue, app_id=form.app_id)
     entry = WorkflowEntry({}, Graph(root_node=node), 0, node.runtime_state)
     for event in list(engine_events(node))[1:]:
-        WorkflowBasedAppRunner.handle_event(runner, entry, event)
-    queue_events = [call.args[0] for call in runner._publish_event.call_args_list]
+        runner.handle_event(entry, event)
+    queue.stop_listen(execution_state=AppExecutionState.TERMINAL)
+    queue_events = [message.event for message in queue.listen()]
     assert isinstance(queue_events[1], QueueNodeSucceededEvent)
     converter = _build_converter()
     converter.workflow_start_to_stream_response(
@@ -370,11 +377,15 @@ def test_file_segments_survive_node_graph_queue_and_response(form: Form) -> None
     assert isinstance(graph_events[2], NodeRunSucceededEvent)
     assert graph_events[1].rendered_content == "Answer: yes; files: [1 files]; attachment: [file]"
     assert isinstance(graph_events[2].node_run_result.outputs["attachment"], FileSegment)
-    runner = create_autospec(WorkflowBasedAppRunner, instance=True)
-    WorkflowBasedAppRunner.handle_event(
-        runner, WorkflowEntry({}, Graph(root_node=node), 0, node.runtime_state), graph_events[1]
+    queue = WorkflowAppQueueManager(
+        task_id="task-1", user_id="user-1", invoke_from=InvokeFrom.DEBUGGER, app_mode="workflow"
     )
-    queue_event = runner._publish_event.call_args.args[0]
+    runner = WorkflowBasedAppRunner(queue_manager=queue, app_id=form.app_id)
+    runner.handle_event(WorkflowEntry({}, Graph(root_node=node), 0, node.runtime_state), graph_events[1])
+    queue.stop_listen(execution_state=AppExecutionState.TERMINAL)
+    [queue_event] = [message.event for message in queue.listen()]
+    assert isinstance(queue_event, QueueHumanInputFormFilledEvent)
+    assert queue_event.submitted_data is not None
     assert isinstance(queue_event.submitted_data["attachment"], FileSegment)
     assert isinstance(queue_event.submitted_data["files"], ArrayFileSegment)
     converter = _build_converter()