fix(api): filter Graphon workflow response events

Apply Graphon ResponseStreamFilter to workflow-based app runners so workflow execution consumes filtered graph events in both streaming and blocking paths.

Pin Graphon to the PR revision while the change waits for the 0.5.0 release, and update unit coverage for the filtered event stream behavior.

Author: laipz8200

api/core/app/apps/advanced_chat/app_runner.py

@@ -246,7 +246,11 @@ def run(self):
         for layer in self._graph_engine_layers:
             workflow_entry.graph_engine.layer(layer)
 
-        generator = workflow_entry.run()
+        generator = self._iter_workflow_events(
+            workflow_entry,
+            workflow_entry.run(),
+            stream=self.application_generate_entity.stream,
+        )
 
         for event in generator:
             self._handle_event(workflow_entry, event)

api/core/app/apps/workflow/app_runner.py

@@ -169,7 +169,11 @@ def run(self):
         for layer in self._graph_engine_layers:
             workflow_entry.graph_engine.layer(layer)
 
-        generator = workflow_entry.run()
+        generator = self._iter_workflow_events(
+            workflow_entry,
+            workflow_entry.run(),
+            stream=self.application_generate_entity.stream,
+        )
 
         for event in generator:
             self._handle_event(workflow_entry, event)

api/core/app/apps/workflow_app_runner.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterable, Mapping, Sequence
 from typing import Any, cast
 
 from pydantic import ValidationError
@@ -51,6 +51,7 @@
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from graphon.entities.graph_config import NodeConfigDictAdapter
 from graphon.entities.pause_reason import HumanInputRequired
+from graphon.filters import GraphEventFilterContext, ResponseStreamFilter, filter_graph_events
 from graphon.graph import Graph
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import (
@@ -381,6 +382,21 @@ def _get_graph_and_variable_pool_for_single_node_run(
 
         return graph, variable_pool
 
+    @staticmethod
+    def _iter_workflow_events(
+        workflow_entry: WorkflowEntry,
+        events: Iterable[GraphEngineEvent],
+        *,
+        stream: bool,
+    ) -> Iterable[GraphEngineEvent]:
+        _ = stream
+
+        return filter_graph_events(
+            events,
+            context=GraphEventFilterContext.from_engine(workflow_entry.graph_engine),
+            filters=[ResponseStreamFilter()],
+        )
+
     @staticmethod
     def _build_agent_strategy_info(event: NodeRunStartedEvent) -> AgentStrategyInfo | None:
         raw_agent_strategy = event.extras.get("agent_strategy")

api/pyproject.toml

@@ -97,6 +97,7 @@ dify-trace-mlflow = { workspace = true }
 dify-trace-opik = { workspace = true }
 dify-trace-tencent = { workspace = true }
 dify-trace-weave = { workspace = true }
+graphon = { git = "https://github.com/langgenius/graphon.git", rev = "853bd461c42489a3fcf26cf36f2831ba03c444a6" }
 
 [tool.uv]
 default-groups = ["storage", "tools", "vdb-all", "trace-all"]

api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_conversation_variables.py

@@ -100,6 +100,7 @@ def test_missing_conversation_variables_are_added(self):
         mock_app_generate_entity.single_iteration_run = None
         mock_app_generate_entity.single_loop_run = None
         mock_app_generate_entity.trace_manager = None
+        mock_app_generate_entity.stream = False
 
         # Create runner
         runner = AdvancedChatAppRunner(
@@ -245,6 +246,7 @@ def test_no_variables_creates_all(self):
         mock_app_generate_entity.single_iteration_run = None
         mock_app_generate_entity.single_loop_run = None
         mock_app_generate_entity.trace_manager = None
+        mock_app_generate_entity.stream = False
 
         # Create runner
         runner = AdvancedChatAppRunner(
@@ -405,6 +407,7 @@ def test_all_variables_exist_no_changes(self):
         mock_app_generate_entity.single_iteration_run = None
         mock_app_generate_entity.single_loop_run = None
         mock_app_generate_entity.trace_manager = None
+        mock_app_generate_entity.stream = False
 
         # Create runner
         runner = AdvancedChatAppRunner(

api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_input_moderation.py

@@ -64,6 +64,7 @@ def build_runner():
     gen.single_iteration_run = None
     gen.single_loop_run = None
     gen.trace_manager = None
+    gen.stream = False
 
     runner = AdvancedChatAppRunner(
         application_generate_entity=gen,

api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py

@@ -234,6 +234,39 @@ def _graph_init(**kwargs):
         assert graph is not None
         assert variable_pool.get(["sys", "conversation_id"]).value == "conv-1"
 
+    @pytest.mark.parametrize("stream", [False, True])
+    def test_iter_workflow_events_filters_response_stream(self, stream: bool):
+        runner = WorkflowBasedAppRunner(queue_manager=SimpleNamespace(), app_id="app")
+        graph_runtime_state = GraphRuntimeState(
+            variable_pool=VariablePool.from_bootstrap(system_variables=default_system_variables()),
+            start_at=0.0,
+        )
+        workflow_entry = SimpleNamespace(
+            graph_engine=SimpleNamespace(
+                graph=SimpleNamespace(nodes={}),
+                graph_runtime_state=graph_runtime_state,
+            )
+        )
+
+        events = iter(
+            [
+                GraphRunStartedEvent(),
+                NodeRunStreamChunkEvent(
+                    id="exec",
+                    node_id="llm",
+                    node_type=BuiltinNodeTypes.LLM,
+                    selector=["llm", "text"],
+                    chunk="raw",
+                    is_final=False,
+                ),
+                GraphRunSucceededEvent(outputs={"answer": "done"}),
+            ]
+        )
+
+        filtered_events = list(runner._iter_workflow_events(workflow_entry, events, stream=stream))
+
+        assert [type(event) for event in filtered_events] == [GraphRunStartedEvent, GraphRunSucceededEvent]
+
     def test_handle_graph_run_events_and_pause_notifications(self, monkeypatch: pytest.MonkeyPatch):
         published: list[object] = []
 

api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py

@@ -53,6 +53,7 @@ def test_run_uses_single_node_execution_branch(
     app_generate_entity.trace_manager = None
     app_generate_entity.single_iteration_run = single_iteration_run
     app_generate_entity.single_loop_run = single_loop_run
+    app_generate_entity.stream = False
 
     workflow = MagicMock(spec=Workflow)
     workflow.tenant_id = "tenant"

api/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.py

@@ -1,3 +1,4 @@
+from graphon.filters import GraphEventFilterContext, ResponseStreamFilter, filter_graph_events
 from graphon.graph_engine import GraphEngine, GraphEngineConfig
 from graphon.graph_engine.command_channels import InMemoryChannel
 from graphon.graph_events import (
@@ -31,7 +32,13 @@ def test_tool_in_chatflow():
         config=GraphEngineConfig(),
     )
 
-    events = list(engine.run())
+    events = list(
+        filter_graph_events(
+            engine.run(),
+            context=GraphEventFilterContext.from_engine(engine),
+            filters=[ResponseStreamFilter()],
+        )
+    )
 
     # Check for successful completion
     success_events = [e for e in events if isinstance(e, GraphRunSucceededEvent)]