fix(api): filter Graphon workflow response events

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 = "a24cf1f227c4b91494779be40c754cb4107c60ed" }
 
 [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)]

packages/contracts/README.md

@@ -8,14 +8,14 @@
 
 Snapshot generated from `packages/contracts/generated/api/readiness.json` after running `pnpm -C packages/contracts gen-api-contract-from-openapi`.
 
-Are we OpenAPI ready? **No.** Current generated API contracts are **16.6% ready**.
+Are we OpenAPI ready? **No.** Current generated API contracts are **16.7% ready**.
 
 | Surface   |   Ready | Not ready |   Total |   Ready % |
 | --------- | ------: | --------: | ------: | --------: |
-| console   |      95 |       475 |     570 |     16.7% |
+| console   |      96 |       474 |     570 |     16.8% |
 | service   |      16 |        72 |      88 |     18.2% |
 | web       |       5 |        36 |      41 |     12.2% |
-| **total** | **116** |   **583** | **699** | **16.6%** |
+| **total** | **117** |   **582** | **699** | **16.7%** |
 
 Readiness here means the generated contract operation is not marked with:
 

packages/contracts/generated/api/console/apps/orpc.gen.ts

@@ -426,16 +426,10 @@ export const imports = {
 
 /**
  * Get workflow online users
- *
- * Generated contract types may be inaccurate because backend OpenAPI annotations are incomplete. Do not migrate callers until the generated contract is accurate.
- *
- * @deprecated
  */
 export const post3 = oc
   .route({
-    deprecated: true,
-    description:
-      'Get workflow online users\n\nGenerated contract types may be inaccurate because backend OpenAPI annotations are incomplete. Do not migrate callers until the generated contract is accurate.',
+    description: 'Get workflow online users',
     inputStructure: 'detailed',
     method: 'POST',
     operationId: 'postAppsWorkflowsOnlineUsers',

packages/contracts/generated/api/readiness.json

@@ -1,7 +1,7 @@
 {
   "surfaces": {
     "console": {
-      "notReady": 475,
+      "notReady": 474,
       "total": 570
     },
     "service": {