diff --git a/api/core/app/apps/advanced_chat/app_runner.py b/api/core/app/apps/advanced_chat/app_runner.py index 4e57b4dedc5b24..bac0a4597efaeb 100644 --- a/api/core/app/apps/advanced_chat/app_runner.py +++ b/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) diff --git a/api/core/app/apps/workflow/app_runner.py b/api/core/app/apps/workflow/app_runner.py index cfb9208486f012..fa4c93bdc19abe 100644 --- a/api/core/app/apps/workflow/app_runner.py +++ b/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) diff --git a/api/core/app/apps/workflow_app_runner.py b/api/core/app/apps/workflow_app_runner.py index 84e9573416de32..79483b7d7a67c0 100644 --- a/api/core/app/apps/workflow_app_runner.py +++ b/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") diff --git a/api/pyproject.toml b/api/pyproject.toml index 3ddc1e30a92d31..0bef694ad38576 100644 --- a/api/pyproject.toml +++ b/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"] diff --git a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_conversation_variables.py b/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_conversation_variables.py index 370f7abb8bce52..10a43848311602 100644 --- a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_conversation_variables.py +++ b/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( diff --git a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_input_moderation.py b/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_input_moderation.py index 5d8faee897163d..075cd2e32b1aaf 100644 --- a/api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_input_moderation.py +++ b/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, diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py index 3949c41eae6fe7..1aa86f677d59ff 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py +++ b/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] = [] diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py index 248fed538836e6..c03fb42ec8a93b 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py +++ b/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" diff --git a/api/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.py b/api/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.py index ba1e74f3e0e6ae..9f785688759c58 100644 --- a/api/tests/unit_tests/core/workflow/graph_engine/test_tool_in_chatflow.py +++ b/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)] diff --git a/api/uv.lock b/api/uv.lock index bb150e8cf3ad06..c54180b6929208 100644 --- a/api/uv.lock +++ b/api/uv.lock @@ -1628,7 +1628,7 @@ requires-dist = [ { name = "gmpy2", specifier = ">=2.3.0" }, { name = "google-api-python-client", specifier = ">=2.196.0" }, { name = "google-cloud-aiplatform", specifier = ">=1.151.0,<2.0.0" }, - { name = "graphon", specifier = "~=0.4.0" }, + { name = "graphon", git = "https://github.com/langgenius/graphon.git?rev=a24cf1f227c4b91494779be40c754cb4107c60ed" }, { name = "gunicorn", specifier = ">=26.0.0" }, { name = "httpx", extras = ["socks"], specifier = ">=0.28.1,<1.0.0" }, { name = "httpx-sse", specifier = "~=0.4.0" }, @@ -2985,7 +2985,7 @@ httpx = [ [[package]] name = "graphon" version = "0.4.0" -source = { registry = "https://pypi.org/simple" } +source = { git = "https://github.com/langgenius/graphon.git?rev=a24cf1f227c4b91494779be40c754cb4107c60ed#a24cf1f227c4b91494779be40c754cb4107c60ed" } dependencies = [ { name = "charset-normalizer" }, { name = "httpx" }, @@ -3005,10 +3005,6 @@ dependencies = [ { name = "unstructured", extra = ["docx", "epub", "md", "ppt", "pptx"] }, { name = "webvtt-py" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/76/24/eb1e7983404dcac84816b76ea450e1bb97023e55e00c699d609340bc361e/graphon-0.4.0.tar.gz", hash = "sha256:afb0c7a58f89e09cfa585296429b4d08cd0df80b9ac54d550f88e7d76ec48ee0", size = 261812, upload-time = "2026-05-13T11:48:39.198Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/b7/de/bad6b3fd1e4b4defc16e6ea106e55c44725a159f1d191a99877bce1c9931/graphon-0.4.0-py3-none-any.whl", hash = "sha256:b33f95886da823d5b1b53d663a4f5f8fa383c37740f3bd19297b8d140fcb804c", size = 372711, upload-time = "2026-05-13T11:48:37.712Z" }, -] [[package]] name = "graphql-core" diff --git a/packages/contracts/README.md b/packages/contracts/README.md index ae682faa539388..e9320c921d66a7 100644 --- a/packages/contracts/README.md +++ b/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: diff --git a/packages/contracts/generated/api/console/apps/orpc.gen.ts b/packages/contracts/generated/api/console/apps/orpc.gen.ts index 2eea4f13c496ab..a1a0faaefd1ac8 100644 --- a/packages/contracts/generated/api/console/apps/orpc.gen.ts +++ b/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', diff --git a/packages/contracts/generated/api/readiness.json b/packages/contracts/generated/api/readiness.json index af86c7646e3a3b..78fed3ccfaf4e1 100644 --- a/packages/contracts/generated/api/readiness.json +++ b/packages/contracts/generated/api/readiness.json @@ -1,7 +1,7 @@ { "surfaces": { "console": { - "notReady": 475, + "notReady": 474, "total": 570 }, "service": {