fix: workflow incorrectly marked as completed while nodes are still executing #39
Conversation
WalkthroughThe PR modifies skip propagation logic in the graph traversal component by adjusting when edges are marked as skipped relative to recursive propagation calls, and changes behavior for taken edges to enqueue and start downstream node execution. Comprehensive unit tests validate the propagation behavior across various edge-state scenarios. Changes
Sequence Diagram(s)sequenceDiagram
participant Caller
participant SkipPropagator
participant Graph
participant GraphStateManager
Caller->>SkipPropagator: propagate_skip_from_edge(edge)
alt Unknown Incoming Edges
SkipPropagator->>SkipPropagator: Stop processing
else All Incoming Edges Skipped
SkipPropagator->>GraphStateManager: mark_node_skipped(downstream)
else Taken Edges Exist
SkipPropagator->>GraphStateManager: enqueue_node(downstream)
SkipPropagator->>GraphStateManager: mark_node_execution_started(downstream)
end
SkipPropagator->>Graph: get_outgoing_edges(downstream)
loop For Each Outgoing Edge
SkipPropagator->>SkipPropagator: propagate_skip_from_edge(outgoing_edge)
SkipPropagator->>GraphStateManager: mark_edge_skipped(outgoing_edge)
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@api/core/workflow/graph_engine/graph_traversal/skip_propagator.py`:
- Around line 61-65: Remove the debug print statement that logs
downstream_node_id before starting execution; in the method that calls
self._state_manager.start_execution(downstream_node_id) and
self._state_manager.enqueue_node(downstream_node_id) (the skip propagator logic
in skip_propagator.py), delete the print(f"Starting execution for node:
{downstream_node_id}") so only the state manager calls remain and no debug
stdout is emitted in production.
- Around line 83-86: The loop in propagate_skip_from_edge marks edges after
recursing which leaves them UNKNOWN during downstream analysis; change the order
so the edge is marked skipped before recursion: in propagate_skip_from_edge's
outgoing_edges loop call self._state_manager.mark_edge_skipped(edge.id) first,
then call self.propagate_skip_from_edge(edge.id), mirroring the pattern used in
skip_branch_paths and ensuring analyze_edge_states on downstream nodes sees the
edge as SKIPPED.
🧹 Nitpick comments (1)
api/tests/unit_tests/core/workflow/graph_engine/graph_traversal/test_skip_propagator.py (1)
115-150: Consider adding mock foranalyze_edge_statesreturn value.The test sets
get_incoming_edges.return_value = []to stop recursion, but doesn't mockanalyze_edge_states.return_value. The recursivepropagate_skip_from_edgecalls will receive aMagicMockinstead of the expected dict structure, which happens to work becauseMagicMockis truthy for thehas_unknowncheck, stopping recursion by coincidence.For explicit test behavior, consider adding:
# After line 135, add: mock_state_manager.analyze_edge_states.return_value = { "has_unknown": True, "has_taken": False, "all_skipped": False, }
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
api/core/workflow/graph_engine/graph_traversal/skip_propagator.pyapi/tests/unit_tests/core/workflow/graph_engine/graph_traversal/__init__.pyapi/tests/unit_tests/core/workflow/graph_engine/graph_traversal/test_skip_propagator.py
🔇 Additional comments (5)
api/tests/unit_tests/core/workflow/graph_engine/graph_traversal/__init__.py (1)
1-1: LGTM!Standard package initialization with appropriate docstring.
api/tests/unit_tests/core/workflow/graph_engine/graph_traversal/test_skip_propagator.py (4)
13-49: LGTM!Test correctly verifies that propagation stops when unknown incoming edges exist, with proper mock setup and assertions.
51-81: LGTM!Test correctly verifies the new behavior where
start_executionandenqueue_nodeare both called when an edge is taken.
186-242: LGTM!Well-designed test with proper side_effect usage to handle different nodes in the chain. The recursive propagation behavior is thoroughly verified.
244-307: LGTM!Good comprehensive test covering all three main branches with proper mock resets between sub-tests.
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
| # Start execution and enqueue node | ||
| print(f"Starting execution for node: {downstream_node_id}") | ||
| self._state_manager.start_execution(downstream_node_id) | ||
| self._state_manager.enqueue_node(downstream_node_id) | ||
| return |
There was a problem hiding this comment.
Remove debug print statement before merging.
The print() statement on line 62 appears to be debug output that should not be in production code. It will pollute logs/stdout during workflow execution.
🧹 Proposed fix
# If any edge is taken, node may still execute
if edge_states["has_taken"]:
# Start execution and enqueue node
- print(f"Starting execution for node: {downstream_node_id}")
self._state_manager.start_execution(downstream_node_id)
self._state_manager.enqueue_node(downstream_node_id)
return📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # Start execution and enqueue node | |
| print(f"Starting execution for node: {downstream_node_id}") | |
| self._state_manager.start_execution(downstream_node_id) | |
| self._state_manager.enqueue_node(downstream_node_id) | |
| return | |
| # Start execution and enqueue node | |
| self._state_manager.start_execution(downstream_node_id) | |
| self._state_manager.enqueue_node(downstream_node_id) | |
| return |
🤖 Prompt for AI Agents
In `@api/core/workflow/graph_engine/graph_traversal/skip_propagator.py` around
lines 61 - 65, Remove the debug print statement that logs downstream_node_id
before starting execution; in the method that calls
self._state_manager.start_execution(downstream_node_id) and
self._state_manager.enqueue_node(downstream_node_id) (the skip propagator logic
in skip_propagator.py), delete the print(f"Starting execution for node:
{downstream_node_id}") so only the state manager calls remain and no debug
stdout is emitted in production.
| for edge in outgoing_edges: | ||
| self._state_manager.mark_edge_skipped(edge.id) | ||
| # Recursively propagate skip | ||
| self.propagate_skip_from_edge(edge.id) | ||
| self._state_manager.mark_edge_skipped(edge.id) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# First, let's examine the skip_propagator.py file to understand the implementation
cat -n api/core/workflow/graph_engine/graph_traversal/skip_propagator.py | head -150Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 3859
🏁 Script executed:
#!/bin/bash
# Look at the specific lines in question (83-86) and surrounding context
cat -n api/core/workflow/graph_engine/graph_traversal/skip_propagator.py | sed -n '50,120p'Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 2240
🏁 Script executed:
#!/bin/bash
# Search for test files related to skip propagation
fd -t f '*test*.py' | xargs rg -l 'skip.*propagat|propagat.*skip' 2>/dev/null | head -10Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 741
🏁 Script executed:
#!/bin/bash
# Let's examine the test file for skip propagator
cat -n api/tests/unit_tests/core/workflow/graph_engine/graph_traversal/test_skip_propagator.pyRepository: qodo-benchmark/dify-combined-coderabbit
Length of output: 13929
🏁 Script executed:
#!/bin/bash
# Look at the graph state manager to understand how analyze_edge_states works
rg -n 'analyze_edge_states|mark_edge_skipped' api/core/workflow/graph_engine/graph_state_manager.py -A 10 | head -80Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 740
🏁 Script executed:
#!/bin/bash
# Check how edge state is initialized and what the default state is
rg -n 'class NodeState|NodeState\.' api/core/workflow/ -A 3 | head -40Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 3014
🏁 Script executed:
#!/bin/bash
# Check initialization of edge states in GraphStateManager
rg -n 'class GraphStateManager|def __init__|edge.*state' api/core/workflow/graph_engine/graph_state_manager.py -B 2 -A 5 | head -80Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 2477
🏁 Script executed:
#!/bin/bash
# Check Edge class initialization to confirm state defaults
rg -n 'class Edge|self\.state.*=.*NodeState' api/core/workflow/graph.py -A 3 | head -40Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 149
🏁 Script executed:
#!/bin/bash
# Look at the git history or comments around this change
cat -n api/core/workflow/graph_engine/graph_traversal/skip_propagator.py | sed -n '70,100p'Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 1268
🏁 Script executed:
#!/bin/bash
# Search for any other test files that might test integration behavior
fd -type f -name '*.py' | xargs grep -l '_propagate_skip_to_node\|skip_propagator' | grep -i testRepository: qodo-benchmark/dify-combined-coderabbit
Length of output: 253
🏁 Script executed:
#!/bin/bash
# Find the Graph class definition
find api -name '*.py' -type f | xargs grep -l 'class Graph' | head -5Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 299
🏁 Script executed:
#!/bin/bash
# Search for Edge class initialization
find api -name '*.py' -type f | xargs grep -l 'class Edge' | head -5Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 233
🏁 Script executed:
#!/bin/bash
# Examine the Edge class
cat -n api/core/workflow/graph/edge.py | head -50Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 674
🏁 Script executed:
#!/bin/bash
# Let me verify this is indeed the problematic scenario by checking if there's a comment in the code explaining the order
cat -n api/core/workflow/graph_engine/graph_traversal/skip_propagator.py | sed -n '71,87p'Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 807
🏁 Script executed:
#!/bin/bash
# Check if there are any integration/end-to-end tests that would actually catch this issue
find api/tests -name '*.py' -type f | xargs grep -l 'propagate_skip\|skip_propagator'Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 170
🏁 Script executed:
#!/bin/bash
# Let me see if there are any comments or documentation about why this order was chosen
git log --oneline -n 20 api/core/workflow/graph_engine/graph_traversal/skip_propagator.py 2>/dev/null || echo "Git not available"Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 101
🏁 Script executed:
#!/bin/bash
# Check if there are any integration tests in the workflow engine that test end-to-end behavior
find api/tests -name '*.py' -type f | xargs grep -l 'graph_engine\|workflow.*execution' | head -5Repository: qodo-benchmark/dify-combined-coderabbit
Length of output: 559
🏁 Script executed:
#!/bin/bash
# Let me create a simple Python script to trace through the logic and confirm the bug
python3 << 'EOF'
from enum import Enum
class NodeState(Enum):
UNKNOWN = "unknown"
SKIPPED = "skipped"
TAKEN = "taken"
def analyze_edge_states(edges):
"""Simulates the analyze_edge_states function"""
states = {edge['state'] for edge in edges}
return {
"has_unknown": NodeState.UNKNOWN in states,
"has_taken": NodeState.TAKEN in states,
"all_skipped": states == {NodeState.SKIPPED} if states else True,
}
# Scenario: propagate_skip_from_edge is called BEFORE mark_edge_skipped
print("=== SCENARIO 1: propagate_skip_from_edge called BEFORE mark_edge_skipped ===")
print("(Current buggy implementation in _propagate_skip_to_node)")
# Edge state is still UNKNOWN when propagate_skip_from_edge is called
incoming_edges = [
{'state': NodeState.UNKNOWN} # Edge being processed - NOT marked yet
]
edge_states = analyze_edge_states(incoming_edges)
print(f"Edge states: {edge_states}")
print(f"has_unknown={edge_states['has_unknown']} -> Early return at line 57")
print(f"Propagation STOPS prematurely\n")
# Scenario 2: mark_edge_skipped is called BEFORE propagate_skip_from_edge
print("=== SCENARIO 2: mark_edge_skipped called BEFORE propagate_skip_from_edge ===")
print("(Correct implementation in skip_branch_paths)")
# Edge state is SKIPPED when propagate_skip_from_edge is called
incoming_edges = [
{'state': NodeState.SKIPPED} # Edge was marked first
]
edge_states = analyze_edge_states(incoming_edges)
print(f"Edge states: {edge_states}")
print(f"all_skipped={edge_states['all_skipped']} -> Continues to propagate skip")
print(f"Propagation continues correctly")
EOFRepository: qodo-benchmark/dify-combined-coderabbit
Length of output: 649
Swap the order: mark edge as skipped before recursive propagation.
At lines 85-86, propagate_skip_from_edge(edge.id) is called before mark_edge_skipped(edge.id). This causes the edge to remain in UNKNOWN state during the recursive call. When propagate_skip_from_edge calls analyze_edge_states on the downstream node's incoming edges, it will see has_unknown=True (because this edge hasn't been marked yet) and return early at line 57, stopping propagation prematurely.
In contrast, skip_branch_paths (lines 95-97) correctly marks the edge first before propagating. Swap the order to match that pattern:
for edge in outgoing_edges:
self._state_manager.mark_edge_skipped(edge.id)
self.propagate_skip_from_edge(edge.id)
🤖 Prompt for AI Agents
In `@api/core/workflow/graph_engine/graph_traversal/skip_propagator.py` around
lines 83 - 86, The loop in propagate_skip_from_edge marks edges after recursing
which leaves them UNKNOWN during downstream analysis; change the order so the
edge is marked skipped before recursion: in propagate_skip_from_edge's
outgoing_edges loop call self._state_manager.mark_edge_skipped(edge.id) first,
then call self.propagate_skip_from_edge(edge.id), mirroring the pattern used in
skip_branch_paths and ensuring analyze_edge_states on downstream nodes sees the
edge as SKIPPED.
Benchmark PR from agentic-review-benchmarks#13
Summary by CodeRabbit
Bug Fixes
Tests
✏️ Tip: You can customize this high-level summary in your review settings.