diff --git a/features/steps/tdd_langgraph_execute_closed_state_steps.py b/features/steps/tdd_langgraph_execute_closed_state_steps.py new file mode 100644 index 000000000..ab710b1ac --- /dev/null +++ b/features/steps/tdd_langgraph_execute_closed_state_steps.py @@ -0,0 +1,95 @@ +"""Behave step definitions for TDD Issue #9994. + +Verifies that LangGraph.execute() raises RuntimeError when StateManager is closed, +preventing silent state corruption after StateManager.close() has been called. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.langgraph.graph import GraphConfig, LangGraph +from cleveragents.langgraph.state import GraphState + + +def _make_graph(name: str = "test-graph") -> LangGraph: + """Create a minimal LangGraph for testing.""" + config = GraphConfig(name=name) + graph = LangGraph(config=config) + graph.is_running = True + graph.stream_router.send_message = MagicMock() + return graph + + +@given("a LangGraph instance in running state with a closed StateManager") +def step_graph_running_with_closed_state_manager(context: Context) -> None: + context.graph = _make_graph("closed-graph") + context.graph.state_manager.close() + + +@when("I attempt to execute the graph after close") +def step_execute_after_close(context: Context) -> None: + context.execute_error = None + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(context.graph.execute({"messages": []})) + except RuntimeError as exc: + context.execute_error = exc + finally: + loop.close() + replacement_loop = asyncio.new_event_loop() + asyncio.set_event_loop(replacement_loop) + + +@then('a RuntimeError should be raised with message "StateManager is closed"') +def step_assert_runtime_error(context: Context) -> None: + assert context.execute_error is not None, ( + "Expected RuntimeError but no error was raised" + ) + assert isinstance(context.execute_error, RuntimeError), ( + f"Expected RuntimeError, got {type(context.execute_error).__name__}" + ) + assert "StateManager is closed" in str(context.execute_error), ( + f'Expected "StateManager is closed" in error message, ' + f"got: {context.execute_error}" + ) + + +@given("a LangGraph instance in running state with an open StateManager") +def step_graph_running_with_open_state_manager(context: Context) -> None: + context.graph = _make_graph("open-graph") + assert not context.graph.state_manager.is_closed + + +@when("I execute the graph with valid input") +def step_execute_with_valid_input(context: Context) -> None: + context.execute_result = None + context.execute_error = None + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + context.execute_result = loop.run_until_complete( + context.graph.execute({"messages": []}) + ) + except Exception as exc: # pylint: disable=broad-except + context.execute_error = exc + finally: + loop.close() + replacement_loop = asyncio.new_event_loop() + asyncio.set_event_loop(replacement_loop) + + +@then("the execution should succeed and return a GraphState") +def step_assert_execution_success(context: Context) -> None: + assert context.execute_error is None, ( + f"Expected no error but got: {context.execute_error}" + ) + assert context.execute_result is not None, "Expected a GraphState result" + assert isinstance(context.execute_result, GraphState), ( + f"Expected GraphState, got {type(context.execute_result).__name__}" + ) diff --git a/features/tdd_langgraph_execute_closed_state.feature b/features/tdd_langgraph_execute_closed_state.feature new file mode 100644 index 000000000..94cff58c4 --- /dev/null +++ b/features/tdd_langgraph_execute_closed_state.feature @@ -0,0 +1,18 @@ +@tdd_issue @tdd_issue_9994 +Feature: TDD Issue #9994 — LangGraph.execute() bypasses is_closed guard + As a developer using LangGraph + I want execute() to raise RuntimeError when StateManager is closed + So that state corruption after close() is prevented + + Background: + Given the state management system is available + + Scenario: execute() raises RuntimeError after StateManager.close() + Given a LangGraph instance in running state with a closed StateManager + When I attempt to execute the graph after close + Then a RuntimeError should be raised with message "StateManager is closed" + + Scenario: execute() succeeds when StateManager is open + Given a LangGraph instance in running state with an open StateManager + When I execute the graph with valid input + Then the execution should succeed and return a GraphState diff --git a/src/cleveragents/langgraph/graph.py b/src/cleveragents/langgraph/graph.py index a1b36ccb3..17fc46583 100644 --- a/src/cleveragents/langgraph/graph.py +++ b/src/cleveragents/langgraph/graph.py @@ -108,7 +108,9 @@ class LangGraph: # pylint: disable=too-many-instance-attributes if isinstance(input_data, GraphState) else GraphState.from_dict(input_data) ) - # Replace state manager state for a fresh execution context + # Use replace_state() so the is_closed guard is enforced and state stream + # subscribers are notified through the proper StateManager API, preventing + # silent state corruption after StateManager.close() has been called. self.state_manager.replace_state(state) start_stream = f"__{self.name}_node_start__" if start_stream in self.stream_router.streams: diff --git a/src/cleveragents/langgraph/state.py b/src/cleveragents/langgraph/state.py index 6d8650e09..ecc7a7605 100644 --- a/src/cleveragents/langgraph/state.py +++ b/src/cleveragents/langgraph/state.py @@ -237,12 +237,16 @@ class StateManager: # pylint: disable=too-many-instance-attributes def replace_state(self, new_state: GraphState) -> None: """Atomically replace the internal state and notify subscribers. + Raises :class:`RuntimeError` if the manager has been closed. + .. note:: See :meth:`update_state` for a note on emission ordering under concurrent access. """ with self._lock: + if self.is_closed: + raise RuntimeError("StateManager is closed") self.state = new_state.model_copy(deep=True) state_copy = new_state.model_copy(deep=True) self.state_stream.on_next(state_copy)