fix: preserve ResponseStreamFilter state across workflow pause/resume#38540
Merged
Conversation
… context Adds a required response_stream_filter parameter to PauseStatePersistenceLayer so the filter's serialized state is dumped into WorkflowResumptionContext on pause and can be restored via get_response_stream_filter() on resume.
…ation test helper
Nothing reads it externally; keeping it private closes off a path for a future caller to silently swap in a second, decoupled filter instance.
Contributor
Pyrefly Type Coverage
|
Contributor
Pyrefly Diffbase → PR--- /tmp/pyrefly_base.txt 2026-07-08 09:10:07.975524118 +0000
+++ /tmp/pyrefly_pr.txt 2026-07-08 09:09:52.421390770 +0000
@@ -614,6 +614,8 @@
--> tests/integration_tests/workflow/test_response_stream_filter_integration.py:74:12
ERROR Object of class `object` has no attribute `strip` [missing-attribute]
--> tests/integration_tests/workflow/test_response_stream_filter_integration.py:75:37
+ERROR Argument `core.workflow.nodes.human_input.entities.HumanInputNodeData` is not assignable to parameter `data` with type `graphon.nodes.human_input.entities.HumanInputNodeData` in function `graphon.nodes.human_input.human_input_node.HumanInputNode.__init__` [bad-argument-type]
+ --> tests/integration_tests/workflow/test_response_stream_filter_pause_resume_integration.py:113:14
ERROR Argument value `Literal[0]` violates Pydantic `ge` constraint `Literal[1]` for field `page` [bad-argument-type]
--> tests/test_containers_integration_tests/controllers/console/app/test_app_apis.py:671:51
ERROR Object of class `RequestError` has no attribute `response` [missing-attribute]
@@ -645,29 +647,29 @@
ERROR Argument `SimpleNamespace` is not assignable to parameter `webapp_settings` with type `WebAppSettings | None` in function `controllers.web.wraps._validate_user_accessibility` [bad-argument-type]
--> tests/test_containers_integration_tests/controllers/web/test_wraps.py:182:33
ERROR Attribute `file_service` of class `TestPauseStatePersistenceLayerTestContainers` is a read-only descriptor with no `__set__` and cannot be set [read-only]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:156:9
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:174:9
ERROR Attribute `workflow_run_service` of class `TestPauseStatePersistenceLayerTestContainers` is a read-only descriptor with no `__set__` and cannot be set [read-only]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:157:9
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:175:9
ERROR Argument `Connection | Engine` is not assignable to parameter `session_factory` with type `Engine | sessionmaker[Session]` in function `core.app.layers.pause_state_persist_layer.PauseStatePersistenceLayer.__init__` [bad-argument-type]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:295:29
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:313:29
ERROR Argument `dict[str, str]` is not assignable to parameter `outputs` with type `dict[str, object] | None` in function `TestPauseStatePersistenceLayerTestContainers._create_graph_runtime_state` [bad-argument-type]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:312:21
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:331:21
ERROR Argument `dict[tuple[str, str], dict[str, str] | str]` is not assignable to parameter `variables` with type `dict[tuple[str, str], object] | None` in function `TestPauseStatePersistenceLayerTestContainers._create_graph_runtime_state` [bad-argument-type]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:315:23
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:334:23
ERROR Argument `dict[tuple[str, str], dict[str, str] | list[int] | str]` is not assignable to parameter `variables` with type `dict[tuple[str, str], object] | None` in function `TestPauseStatePersistenceLayerTestContainers._create_graph_runtime_state` [bad-argument-type]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:379:23
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:398:23
ERROR Object of class `FixtureFunctionDefinition` has no attribute `_workflow_run_repo` [missing-attribute]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:391:24
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:410:24
ERROR Object of class `AdvancedChatAppGenerateEntity` has no attribute `workflow_execution_id` [missing-attribute]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:405:16
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:424:16
ERROR Argument `dict[str, LiteralString]` is not assignable to parameter `outputs` with type `dict[str, object] | None` in function `TestPauseStatePersistenceLayerTestContainers._create_graph_runtime_state` [bad-argument-type]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:446:21
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:465:21
ERROR Object of class `AdvancedChatAppGenerateEntity` has no attribute `workflow_execution_id` [missing-attribute]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:470:16
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:489:16
ERROR Object of class `FixtureFunctionDefinition` has no attribute `_workflow_run_repo` [missing-attribute]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:532:24
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:551:24
ERROR Object of class `AdvancedChatAppGenerateEntity` has no attribute `workflow_execution_id` [missing-attribute]
- --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:535:16
+ --> tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py:554:16
ERROR Object of class `NoneType` has no attribute `id` [missing-attribute]
--> tests/test_containers_integration_tests/core/rag/retrieval/test_dataset_retrieval_integration.py:38:23
ERROR Object of class `NoneType` has no attribute `id` [missing-attribute]
@@ -6599,39 +6601,41 @@
ERROR Argument `TestWorkflowEntry.test_single_step_run_injects_code_limits.StubWorkflow` is not assignable to parameter `workflow` with type `Workflow` in function `core.workflow.workflow_entry.WorkflowEntry.single_step_run` [bad-argument-type]
--> tests/unit_tests/core/workflow/test_workflow_entry.py:145:22
ERROR Object of class `_FakeModelNodeMixin` has no attribute `node_data` [missing-attribute]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:73:35
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:74:35
ERROR Object of class `_FakeModelNodeMixin` has no attribute `node_data` [missing-attribute]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:74:31
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:75:31
ERROR Argument `SimpleNamespace` is not assignable to parameter `graph_init_params` with type `GraphInitParams` in function `core.workflow.workflow_entry._WorkflowChildEngineBuilder.build_child_engine` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:115:39
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:116:39
ERROR Argument `SimpleNamespace` is not assignable to parameter `parent_graph_runtime_state` with type `GraphRuntimeState` in function `core.workflow.workflow_entry._WorkflowChildEngineBuilder.build_child_engine` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:116:48
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:117:48
ERROR Argument `SimpleNamespace` is not assignable to parameter `graph_init_params` with type `GraphInitParams` in function `core.workflow.workflow_entry._WorkflowChildEngineBuilder.build_child_engine` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:147:35
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:148:35
ERROR Argument `SimpleNamespace` is not assignable to parameter `parent_graph_runtime_state` with type `GraphRuntimeState` in function `core.workflow.workflow_entry._WorkflowChildEngineBuilder.build_child_engine` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:148:44
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:149:44
ERROR Argument `SimpleNamespace` is not assignable to parameter `parent_graph_runtime_state` with type `GraphRuntimeState` in function `core.workflow.workflow_entry._WorkflowChildEngineBuilder.build_child_engine` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:225:44
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:226:44
+ERROR Argument `SimpleNamespace` is not assignable to parameter `graph_runtime_state` with type `GraphRuntimeState` in function `core.workflow.workflow_entry.WorkflowEntry.__init__` [bad-argument-type]
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:271:29
ERROR Argument `SimpleNamespace` is not assignable to parameter `graph_runtime_state` with type `GraphRuntimeState` in function `core.workflow.workflow_entry.WorkflowEntry.__init__` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:299:37
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:331:37
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `core.workflow.workflow_entry.WorkflowEntry.single_step_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:461:26
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:507:26
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `core.workflow.workflow_entry.WorkflowEntry.single_step_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:516:26
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:562:26
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `core.workflow.workflow_entry.WorkflowEntry.single_step_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:585:26
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:631:26
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `core.workflow.workflow_entry.WorkflowEntry.single_step_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:643:30
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:689:30
ERROR Class member `EmptySplitKey.split` overrides parent class `UserString` in an inconsistent manner [bad-override]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:866:17
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:912:17
ERROR Class member `EmptySplitKey.split` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:866:17
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:912:17
ERROR Argument `dict[TestMappingUserInputsBranches.test_rejects_invalid_node_variable_key.EmptySplitKey, list[str]]` is not assignable to parameter `variable_mapping` with type `Mapping[str, Sequence[str]]` in function `core.workflow.workflow_entry.WorkflowEntry.mapping_user_inputs_to_variable_pool` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:871:34
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:917:34
ERROR Argument `TestWorkflowEntryTracing.test_traced_node_run_reports_success.FakeNode` is not assignable to parameter `node` with type `Node[Unknown]` in function `core.workflow.workflow_entry.WorkflowEntry._traced_node_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:922:73
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:968:73
ERROR Argument `TestWorkflowEntryTracing.test_traced_node_run_reports_errors.FakeNode` is not assignable to parameter `node` with type `Node[Unknown]` in function `core.workflow.workflow_entry.WorkflowEntry._traced_node_run` [bad-argument-type]
- --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:945:68
+ --> tests/unit_tests/core/workflow/test_workflow_entry_helpers.py:991:68
ERROR Missing argument `case` in function `enterprise.telemetry.contracts.TelemetryEnvelope.__init__` [missing-argument]
--> tests/unit_tests/enterprise/telemetry/test_contracts.py:111:30
ERROR Missing argument `tenant_id` in function `enterprise.telemetry.contracts.TelemetryEnvelope.__init__` [missing-argument]
@@ -8413,19 +8417,19 @@
ERROR Argument `SimpleNamespace` is not assignable to parameter `conversation` with type `Conversation` in function `tasks.app_generate.workflow_execute_task._resume_advanced_chat` [bad-argument-type]
--> tests/unit_tests/tasks/test_workflow_execute_task.py:722:22
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow_run` with type `WorkflowRun` in function `tasks.app_generate.workflow_execute_task._resume_advanced_chat` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:729:22
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:730:22
ERROR Argument `SimpleNamespace` is not assignable to parameter `app_model` with type `App` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:772:19
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:773:19
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:773:18
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:774:18
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow_run` with type `WorkflowRun` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:780:22
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:782:22
ERROR Argument `SimpleNamespace` is not assignable to parameter `app_model` with type `App` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:827:19
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:829:19
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow` with type `Workflow` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:828:18
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:830:18
ERROR Argument `SimpleNamespace` is not assignable to parameter `workflow_run` with type `WorkflowRun` in function `tasks.app_generate.workflow_execute_task._resume_workflow` [bad-argument-type]
- --> tests/unit_tests/tasks/test_workflow_execute_task.py:835:22
+ --> tests/unit_tests/tasks/test_workflow_execute_task.py:838:22
ERROR Object of class `BlobChunkMessage` has no attribute `variable_name`
ERROR Object of class `BlobChunkMessage` has no attribute `variable_name`
ERROR Object of class `BlobChunkMessage` has no attribute `variable_value`
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #38540 +/- ##
==========================================
- Coverage 85.34% 85.33% -0.02%
==========================================
Files 5013 5006 -7
Lines 265940 265610 -330
Branches 50558 50496 -62
==========================================
- Hits 226978 226652 -326
+ Misses 34509 34505 -4
Partials 4453 4453
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
QuantumGhost
approved these changes
Jul 9, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes #38525. After a
human_inputnode pauses a workflow and the run is resumed, the live SSEmessagestream was broken for answer nodes downstream of a branch (if-else) resolved before the pause — even though the finaloutputs.answerwas correct.Root cause:
graphon.filters.ResponseStreamFiltertracks which branch edges have been taken so it knows when downstream answer nodes may stream. It was never persisted across the pause/resume boundary, so a fresh filter on resume had no memory of branch resolution that happened before the pause, and those downstream answer nodes never unlocked for streaming.Fix: thread one
ResponseStreamFilterinstance per run — created (or.loads()-restored) at the pointPauseStatePersistenceLayeris built — through the whole call chain:WorkflowEntry/iter_dify_graph_engine_events→PauseStatePersistenceLayer(now requires the filter, persists its state on pause) → both app runners → both app generators → both resume entry points (UI-driven human-input resume and the trigger/webhook async resume path), restored via a newWorkflowResumptionContext.get_response_stream_filter()accessor.No changes to the
graphonpackage — itsdumps()/loads()/initialize()already worked correctly; the gap was entirely missing dify/api wiring. Backward compatible: the new persisted field is optional, so a workflow run already paused before this ships resumes with today's behavior for that one stale run.DSL For Reproducing Issue
debug.yml
Test plan
tests/integration_tests/workflow/test_response_stream_filter_pause_resume_integration.pyWorkflowEntry, persistence/restoration onPauseStatePersistenceLayer)pytest tests/unit_tests)tests/integration_tests/workflowgreen (pre-existing unrelated failures only: JS code-executor tests requiring a sandbox service not available in this environment)