BUG-HUNT: [error-handling] `LangGraph.execute()` returns immediately without awaiting graph node processing — callers always receive the unprocessed input state #6509

New issue

Open

opened 2026-04-09 21:13:14 +00:00 by HAL9000 · 0 comments

HAL9000 commented

2026-04-09 21:13:14 +00:00

Owner

Bug Report: [error-handling] — `LangGraph.execute()` Returns Without Running Graph Nodes

Severity Assessment

Impact: Every caller of LangGraph.execute() receives the unprocessed input state back, not the result of running the graph. Core graph execution is silently broken — all agents and transforms in the graph are never applied.
Likelihood: 100% — every call to execute() is affected.
Priority: Critical

Location

File: src/cleveragents/langgraph/graph.py
Class: LangGraph
Method: execute
Lines: ~70–85

Description

LangGraph.execute() is an async method but it is not awaiting any of the actual graph node processing. It fires a message to the start stream via send_message (which is fire-and-forget) and then immediately returns the state that was just set — the input state, unchanged.

Evidence

async def execute(self, input_data: GraphState | dict[str, Any]) -> GraphState:
    state = (
        input_data
        if isinstance(input_data, GraphState)
        else GraphState.from_dict(cast(dict[str, Any], input_data))
    )
    # Replace state manager state for a fresh execution context
    self.state_manager.state = state
    self.state_manager.state_stream.on_next(state)
    start_stream = f"__{self.name}_node_start__"
    if start_stream in self.stream_router.streams:
        self.stream_router.send_message(start_stream, state)  # fire-and-forget
    return self.state_manager.get_state()  # <-- returns immediately, state is unchanged!

stream_router.send_message(start_stream, state) triggers reactive stream processing asynchronously. The execute() method returns immediately after calling it, before any node has had a chance to run. self.state_manager.get_state() at that point still holds the original input state (which was just assigned on the line self.state_manager.state = state).

This is confirmed by the _register_node_executor code in the same file which uses concurrent.futures.ThreadPoolExecutor to run nodes — this runs in a background thread that is never awaited by execute().

Expected Behavior

execute() should await the completion of all graph nodes in topological order before returning the final GraphState. The returned state should reflect all transformations applied by the graph's nodes.

Actual Behavior

execute() always returns the input state unchanged. No graph nodes are run from the perspective of the execute() caller.

Suggested Fix

Implement proper sequential or parallel node execution inside execute() using the already-computed _topological_levels() to drive async node dispatch, awaiting all futures at each level before advancing. Example approach:

async def execute(self, input_data: GraphState | dict[str, Any]) -> GraphState:
    state = ...  # as today
    self.state_manager.state = state
    levels = self._topological_levels()
    for level_idx in sorted(levels):
        node_names = levels[level_idx] - {"start", "end"}
        if not node_names:
            continue
        tasks = [node.execute(self.state_manager.get_state()) for n in node_names
                 if (node := self.nodes.get(n)) is not None]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for updates in results:
            if isinstance(updates, dict):
                self.state_manager.update_state(updates)
    return self.state_manager.get_state()

TDD Note

After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: @tdd_issue, @tdd_issue_<this-issue-number>, and @tdd_expected_fail to prove the bug exists before fixing it.

Automated by CleverAgents Bot
Supervisor: Bug Hunting | Agent: bug-hunter

## Bug Report: [error-handling] — `LangGraph.execute()` Returns Without Running Graph Nodes ### Severity Assessment - **Impact**: Every caller of `LangGraph.execute()` receives the unprocessed input state back, not the result of running the graph. Core graph execution is silently broken — all agents and transforms in the graph are never applied. - **Likelihood**: 100% — every call to `execute()` is affected. - **Priority**: Critical ### Location - **File**: `src/cleveragents/langgraph/graph.py` - **Class**: `LangGraph` - **Method**: `execute` - **Lines**: ~70–85 ### Description `LangGraph.execute()` is an `async` method but it is **not awaiting** any of the actual graph node processing. It fires a message to the start stream via `send_message` (which is fire-and-forget) and then **immediately returns** the state that was just set — the input state, unchanged. ### Evidence ```python async def execute(self, input_data: GraphState | dict[str, Any]) -> GraphState: state = ( input_data if isinstance(input_data, GraphState) else GraphState.from_dict(cast(dict[str, Any], input_data)) ) # Replace state manager state for a fresh execution context self.state_manager.state = state self.state_manager.state_stream.on_next(state) start_stream = f"__{self.name}_node_start__" if start_stream in self.stream_router.streams: self.stream_router.send_message(start_stream, state) # fire-and-forget return self.state_manager.get_state() # <-- returns immediately, state is unchanged! ``` `stream_router.send_message(start_stream, state)` triggers reactive stream processing **asynchronously**. The `execute()` method returns immediately after calling it, before any node has had a chance to run. `self.state_manager.get_state()` at that point still holds the original input `state` (which was just assigned on the line `self.state_manager.state = state`). This is confirmed by the `_register_node_executor` code in the same file which uses `concurrent.futures.ThreadPoolExecutor` to run nodes — this runs in a background thread that is never awaited by `execute()`. ### Expected Behavior `execute()` should await the completion of all graph nodes in topological order before returning the final `GraphState`. The returned state should reflect all transformations applied by the graph's nodes. ### Actual Behavior `execute()` always returns the input state unchanged. No graph nodes are run from the perspective of the `execute()` caller. ### Suggested Fix Implement proper sequential or parallel node execution inside `execute()` using the already-computed `_topological_levels()` to drive async node dispatch, awaiting all futures at each level before advancing. Example approach: ```python async def execute(self, input_data: GraphState | dict[str, Any]) -> GraphState: state = ... # as today self.state_manager.state = state levels = self._topological_levels() for level_idx in sorted(levels): node_names = levels[level_idx] - {"start", "end"} if not node_names: continue tasks = [node.execute(self.state_manager.get_state()) for n in node_names if (node := self.nodes.get(n)) is not None] results = await asyncio.gather(*tasks, return_exceptions=True) for updates in results: if isinstance(updates, dict): self.state_manager.update_state(updates) return self.state_manager.get_state() ``` ### Category `error-handling` ### TDD Note After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: `@tdd_issue`, `@tdd_issue_<this-issue-number>`, and `@tdd_expected_fail` to prove the bug exists before fixing it. --- **Automated by CleverAgents Bot** Supervisor: Bug Hunting | Agent: bug-hunter

HAL9000 added the

labels

2026-04-09 21:31:11 +00:00

HAL9000 referenced this issue

2026-04-09 22:20:39 +00:00

[AUTO-BUG-SUP] Bug Hunting Pool Status (Cycle 1) #6373