UAT: LangGraph.execute() does not execute graph nodes — returns initial state without processing #1822

New issue

Open

opened 2026-04-02 23:55:43 +00:00 by freemo · 1 comment

freemo commented

2026-04-02 23:55:43 +00:00

Owner

Metadata

Branch: fix/m6-langgraph-execute-node-processing
Commit Message: fix(langgraph): await async node execution pipeline in LangGraph.execute()
Milestone: v3.5.0
Parent Epic: #390

Background and Context

The LangGraph.execute() method in src/cleveragents/langgraph/graph.py is the primary API for running a LangGraph graph. It is responsible for accepting input data, initialising the execution state, and driving all graph nodes to completion before returning the final state.

This bug was discovered during UAT testing of the LangGraph integration feature area by UAT tester instance uat-tester-3994408-1775170787.

Current Behavior (Bug)

LangGraph.execute() sets the state, sends a message to the start stream, and immediately returns the current state — without waiting for any nodes to process the input. As a result, calling await graph.execute(input_data) returns the input state unchanged, with no node processing having occurred.

File: src/cleveragents/langgraph/graph.py
Lines: 68–78

Buggy code:

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)
    return self.state_manager.get_state()  # Returns BEFORE nodes execute

Root cause: The execute() method sends a message to the start stream (which triggers async stream processing via RxPy) but immediately returns self.state_manager.get_state() without awaiting the completion of the stream-based node execution pipeline. The node execution is event-driven and happens asynchronously, but execute() doesn't await it.

Steps to reproduce:

import asyncio
from cleveragents.langgraph.graph import GraphConfig, LangGraph
from cleveragents.langgraph.nodes import NodeConfig, NodeType, Edge
from cleveragents.agents.base import Agent

class TrackingAgent(Agent):
    def __init__(self):
        super().__init__(name='tracking_agent')
        self.called = False

    async def process_message(self, message, context=None):
        self.called = True
        return f'processed: {message}'

    def get_capabilities(self):
        return ['test']

async def test():
    agent = TrackingAgent()
    config = GraphConfig(
        name='test_exec',
        nodes={
            'agent_node': NodeConfig(name='agent_node', type=NodeType.AGENT, agent='tracking_agent'),
        },
        edges=[
            Edge(source='start', target='agent_node'),
            Edge(source='agent_node', target='end'),
        ],
        entry_point='start'
    )
    graph = LangGraph(config=config, agents={'tracking_agent': agent})
    result = await graph.execute({'messages': [{'role': 'user', 'content': 'hello'}]})

    print(f'Agent was called: {agent.called}')          # Expected: True,  Actual: False
    print(f'Result messages: {result.messages}')         # Expected: processed response, Actual: original input
    print(f'Execution history: {graph.get_execution_history()}')  # Expected: [start, agent_node, end], Actual: [start]

asyncio.run(test())

Expected Behavior

graph.execute() should execute all nodes in the graph in topological order (start → agent_node → end) and return the final state after all nodes have processed the input.

Acceptance Criteria

LangGraph.execute() awaits completion of the full node execution pipeline before returning
All graph nodes are executed in topological order (start → intermediate nodes → end)
The returned GraphState reflects the output of the last node, not the original input
graph.get_execution_history() contains all executed nodes after execute() returns
The fix does not break streaming / reactive use-cases that rely on the RxPy stream router
Existing tests pass; new regression tests are added covering the above criteria

Subtasks

Investigate the RxPy stream-based execution pipeline to understand how node completion can be awaited
Implement an async completion mechanism in LangGraph.execute() (e.g., asyncio.Event, asyncio.Future, or observable completion signal) so that the method awaits full pipeline execution
Ensure the fix handles error propagation correctly (exceptions in nodes surface to the caller)
Tests (Behave): Add BDD scenarios covering sequential node execution, multi-node graphs, and error propagation
Tests (Robot): Add integration test verifying end-to-end graph execution returns final state
Verify coverage ≥ 97% via nox -s coverage_report
Run nox (all default sessions), fix any errors

Definition of Done

This issue is complete when:

All subtasks above are completed and checked off.
A Git commit is created where the first line of the commit message matches the Commit Message in Metadata exactly (fix(langgraph): await async node execution pipeline in LangGraph.execute()), followed by a blank line, then additional lines providing relevant details about the implementation.
The commit is pushed to the remote on the branch matching the Branch in Metadata exactly (fix/m6-langgraph-execute-node-processing).
The commit is submitted as a pull request to master, reviewed, and merged before this issue is marked done.
All nox stages pass.
Coverage >= 97%.

Automated by CleverAgents Bot
Supervisor: UAT Testing | Agent: ca-new-issue-creator

## Metadata - **Branch**: `fix/m6-langgraph-execute-node-processing` - **Commit Message**: `fix(langgraph): await async node execution pipeline in LangGraph.execute()` - **Milestone**: v3.5.0 - **Parent Epic**: #390 ## Background and Context The `LangGraph.execute()` method in `src/cleveragents/langgraph/graph.py` is the primary API for running a LangGraph graph. It is responsible for accepting input data, initialising the execution state, and driving all graph nodes to completion before returning the final state. This bug was discovered during UAT testing of the LangGraph integration feature area by UAT tester instance `uat-tester-3994408-1775170787`. ## Current Behavior (Bug) `LangGraph.execute()` sets the state, sends a message to the start stream, and **immediately returns the current state** — without waiting for any nodes to process the input. As a result, calling `await graph.execute(input_data)` returns the input state unchanged, with no node processing having occurred. **File**: `src/cleveragents/langgraph/graph.py` **Lines**: 68–78 **Buggy code**: ```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) return self.state_manager.get_state() # Returns BEFORE nodes execute ``` **Root cause**: The `execute()` method sends a message to the start stream (which triggers async stream processing via RxPy) but immediately returns `self.state_manager.get_state()` without awaiting the completion of the stream-based node execution pipeline. The node execution is event-driven and happens asynchronously, but `execute()` doesn't await it. **Steps to reproduce**: ```python import asyncio from cleveragents.langgraph.graph import GraphConfig, LangGraph from cleveragents.langgraph.nodes import NodeConfig, NodeType, Edge from cleveragents.agents.base import Agent class TrackingAgent(Agent): def __init__(self): super().__init__(name='tracking_agent') self.called = False async def process_message(self, message, context=None): self.called = True return f'processed: {message}' def get_capabilities(self): return ['test'] async def test(): agent = TrackingAgent() config = GraphConfig( name='test_exec', nodes={ 'agent_node': NodeConfig(name='agent_node', type=NodeType.AGENT, agent='tracking_agent'), }, edges=[ Edge(source='start', target='agent_node'), Edge(source='agent_node', target='end'), ], entry_point='start' ) graph = LangGraph(config=config, agents={'tracking_agent': agent}) result = await graph.execute({'messages': [{'role': 'user', 'content': 'hello'}]}) print(f'Agent was called: {agent.called}') # Expected: True, Actual: False print(f'Result messages: {result.messages}') # Expected: processed response, Actual: original input print(f'Execution history: {graph.get_execution_history()}') # Expected: [start, agent_node, end], Actual: [start] asyncio.run(test()) ``` ## Expected Behavior `graph.execute()` should execute all nodes in the graph in topological order (start → agent_node → end) and return the **final state** after all nodes have processed the input. ## Acceptance Criteria - [ ] `LangGraph.execute()` awaits completion of the full node execution pipeline before returning - [ ] All graph nodes are executed in topological order (start → intermediate nodes → end) - [ ] The returned `GraphState` reflects the output of the last node, not the original input - [ ] `graph.get_execution_history()` contains all executed nodes after `execute()` returns - [ ] The fix does not break streaming / reactive use-cases that rely on the RxPy stream router - [ ] Existing tests pass; new regression tests are added covering the above criteria ## Subtasks - [ ] Investigate the RxPy stream-based execution pipeline to understand how node completion can be awaited - [ ] Implement an async completion mechanism in `LangGraph.execute()` (e.g., `asyncio.Event`, `asyncio.Future`, or observable completion signal) so that the method awaits full pipeline execution - [ ] Ensure the fix handles error propagation correctly (exceptions in nodes surface to the caller) - [ ] Tests (Behave): Add BDD scenarios covering sequential node execution, multi-node graphs, and error propagation - [ ] Tests (Robot): Add integration test verifying end-to-end graph execution returns final state - [ ] Verify coverage ≥ 97% via `nox -s coverage_report` - [ ] Run `nox` (all default sessions), fix any errors ## Definition of Done This issue is complete when: - All subtasks above are completed and checked off. - A Git commit is created where the **first line** of the commit message matches the Commit Message in Metadata exactly (`fix(langgraph): await async node execution pipeline in LangGraph.execute()`), followed by a blank line, then additional lines providing relevant details about the implementation. - The commit is pushed to the remote on the branch matching the **Branch** in Metadata exactly (`fix/m6-langgraph-execute-node-processing`). - The commit is submitted as a **pull request** to `master`, reviewed, and **merged** before this issue is marked done. - All nox stages pass. - Coverage >= 97%. --- **Automated by CleverAgents Bot** Supervisor: UAT Testing | Agent: ca-new-issue-creator

freemo added this to the v3.5.0 milestone

2026-04-02 23:56:11 +00:00

freemo commented

2026-04-02 23:57:34 +00:00

Author

Owner

Issue triaged by project owner:

State: Verified
Priority: Priority/Critical
MoSCoW: MoSCoW/Must Have — LangGraph.execute() not executing graph nodes is a fundamental correctness bug in the core execution engine. Must Have.

Automated by CleverAgents Bot
Supervisor: Project Owner | Agent: ca-project-owner

Issue triaged by project owner: - **State**: Verified - **Priority**: Priority/Critical - **MoSCoW**: MoSCoW/Must Have — LangGraph.execute() not executing graph nodes is a fundamental correctness bug in the core execution engine. Must Have. --- **Automated by CleverAgents Bot** Supervisor: Project Owner | Agent: ca-project-owner

freemo added the

MoSCoW

Must have

label

2026-04-02 23:57:34 +00:00

freemo referenced this issue

2026-04-02 23:58:16 +00:00

[UAT-SUPERVISOR] Progress Report — Cycle 461 — v3.7.0 #1850

freemo added a new dependency

2026-04-03 00:01:44 +00:00

#390 Legendary: Autonomy Hardening & Server Infrastructure (M6)

freemo removed a dependency