features/steps/tdd_langgraph_disposables_steps.py

@@ -0,0 +1,162 @@
+"""Step definitions for TDD Issue #10398.
+
+Verifies that LangGraph._setup_node_stream_subscriptions stores the
+Disposable returned by observable.subscribe() and that stop() disposes
+all stored subscriptions, preventing resource leaks.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+
+from cleveragents.langgraph.graph import GraphConfig, LangGraph
+from cleveragents.langgraph.nodes import NodeConfig, NodeType
+
+
+def _fresh_graph(config: GraphConfig) -> LangGraph:
+    """Construct a LangGraph with isolated scheduler and router."""
+    return LangGraph(config=config, stream_router=None, scheduler=None)
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("the LangGraph module is available")
+def step_langgraph_module_available(_context: object) -> None:
+    """Verify the LangGraph class can be imported (import already at top)."""
+
+
+# ---------------------------------------------------------------------------
+# Scenario: LangGraph initialises an empty subscriptions list
+# ---------------------------------------------------------------------------
+
+
+@when("I construct a LangGraph with default configuration")
+def step_construct_default_graph(context: object) -> None:
+    context.graph = _fresh_graph(GraphConfig(name="tdd-10398-default"))
+
+
+@then("the graph should have an empty _subscriptions list")
+def step_assert_subscriptions_attr_exists(context: object) -> None:
+    # The attribute must exist and be a list (it will be populated by
+    # _setup_node_stream_subscriptions, so it won't be empty after __init__,
+    # but the attribute itself must be a list).
+    assert hasattr(context.graph, "_subscriptions"), (
+        "LangGraph must have a _subscriptions attribute"
+    )
+    assert isinstance(context.graph._subscriptions, list), (
+        "_subscriptions must be a list"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario: _setup_node_stream_subscriptions stores Disposables
+# ---------------------------------------------------------------------------
+
+
+@given("a LangGraph instance with one worker node")
+def step_graph_with_worker_node(context: object) -> None:
+    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
+    context.graph = _fresh_graph(GraphConfig(name="tdd-10398-worker", nodes=nodes))
+
+
+@when("I inspect the subscriptions after construction")
+def step_inspect_subscriptions(_context: object) -> None:
+    """No action needed — subscriptions are populated during __init__."""
+
+
+@then("the _subscriptions list should contain at least one entry per node")
+def step_assert_subscriptions_populated(context: object) -> None:
+    # The graph has start, end, and worker nodes — each should have a
+    # subscription stored.
+    subs = context.graph._subscriptions
+    assert len(subs) >= len(context.graph.nodes), (
+        f"Expected at least {len(context.graph.nodes)} subscriptions, got {len(subs)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario: stop() disposes all stored subscriptions
+# ---------------------------------------------------------------------------
+
+
+@given("a LangGraph instance with tracked disposable subscriptions")
+def step_graph_with_tracked_disposables(context: object) -> None:
+    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
+    context.graph = _fresh_graph(GraphConfig(name="tdd-10398-tracked", nodes=nodes))
+
+    # Replace the real disposables with mocks so we can assert dispose() calls.
+    context.mock_disposables: list[MagicMock] = []
+    for _i in range(len(context.graph._subscriptions)):
+        mock_disp = MagicMock()
+        context.mock_disposables.append(mock_disp)
+    context.graph._subscriptions = list(context.mock_disposables)
+
+
+@when("I call stop() on the graph")
+def step_call_stop(context: object) -> None:
+    context.stop_error: Exception | None = None
+    try:
+        context.graph.stop()
+    except Exception as exc:  # pylint: disable=broad-except
+        context.stop_error = exc
+
+
+@then("every tracked disposable should have been disposed")
+def step_assert_all_disposed(context: object) -> None:
+    assert context.stop_error is None, (
+        f"stop() raised unexpectedly: {context.stop_error}"
+    )
+    for i, mock_disp in enumerate(context.mock_disposables):
+        (
+            mock_disp.dispose.assert_called_once(),
+            (f"Disposable #{i} was not disposed by stop()"),
+        )
+
+
+@then("the _subscriptions list should be empty after stop")
+def step_assert_subscriptions_cleared(context: object) -> None:
+    assert context.graph._subscriptions == [], (
+        "_subscriptions should be cleared after stop()"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario: stop() clears subscriptions even when dispose raises
+# ---------------------------------------------------------------------------
+
+
+@given("a LangGraph instance with a subscription that raises on dispose")
+def step_graph_with_failing_disposable(context: object) -> None:
+    context.graph = _fresh_graph(GraphConfig(name="tdd-10398-fail-disp"))
+
+    failing_disp = MagicMock()
+    failing_disp.dispose.side_effect = RuntimeError("dispose failed")
+    context.graph._subscriptions = [failing_disp]
+
+
+@when("I call stop() on the graph with a failing disposable")
+def step_call_stop_with_failing_disposable(context: object) -> None:
+    context.stop_error = None
+    try:
+        context.graph.stop()
+    except Exception as exc:  # pylint: disable=broad-except
+        context.stop_error = exc
+
+
+@then("stop() should complete without raising an exception")
+def step_assert_stop_no_exception(context: object) -> None:
+    assert context.stop_error is None, (
+        f"stop() raised unexpectedly when a disposable failed: {context.stop_error}"
+    )
+
+
+@then("the _subscriptions list should be empty after the failing stop")
+def step_assert_subscriptions_cleared_after_fail(context: object) -> None:
+    assert context.graph._subscriptions == [], (
+        "_subscriptions should be cleared even when dispose raises"
+    )

features/tdd_langgraph_disposables.feature

@@ -0,0 +1,29 @@
+@tdd_issue @tdd_issue_10398
+Feature: TDD Issue #10398 — LangGraph._setup_node_stream_subscriptions stores and disposes RxPy Disposables
+  As a CleverAgents developer
+  I want LangGraph to store the Disposable returned by observable.subscribe()
+  So that stop() can dispose all subscriptions and prevent resource leaks
+
+  Background:
+    Given the LangGraph module is available
+
+  Scenario: LangGraph initialises an empty subscriptions list
+    When I construct a LangGraph with default configuration
+    Then the graph should have an empty _subscriptions list
+
+  Scenario: _setup_node_stream_subscriptions stores Disposables
+    Given a LangGraph instance with one worker node
+    When I inspect the subscriptions after construction
+    Then the _subscriptions list should contain at least one entry per node
+
+  Scenario: stop() disposes all stored subscriptions
+    Given a LangGraph instance with tracked disposable subscriptions
+    When I call stop() on the graph
+    Then every tracked disposable should have been disposed
+    And the _subscriptions list should be empty after stop
+
+  Scenario: stop() clears subscriptions even when dispose raises
+    Given a LangGraph instance with a subscription that raises on dispose
+    When I call stop() on the graph with a failing disposable
+    Then stop() should complete without raising an exception
+    And the _subscriptions list should be empty after the failing stop

src/cleveragents/langgraph/graph.py

@@ -90,6 +90,8 @@ class LangGraph:  # pylint: disable=too-many-instance-attributes
         self.execution_history: collections.deque[str] = collections.deque(
             maxlen=MAX_EXECUTION_HISTORY
         )
+        # Disposables returned by observable.subscribe(); disposed in stop().
+        self._subscriptions: list[Any] = []
         self.is_running = False
         self._create_graph_streams()
         self._analyze_graph()
@@ -217,7 +219,6 @@ class LangGraph:  # pylint: disable=too-many-instance-attributes
         return on_error
 
     def _setup_node_stream_subscriptions(self) -> None:
-        # Intentionally no-op: stream completion requires no cleanup.
         def on_completed() -> None:
             pass
 
@@ -235,7 +236,8 @@ class LangGraph:  # pylint: disable=too-many-instance-attributes
                 on_error=self._make_on_error_handler(stream_name),
                 on_completed=on_completed,
             )
-            observable.subscribe(observer)
+            disposable = observable.subscribe(observer)
+            self._subscriptions.append(disposable)
 
     def _register_node_executor(self, node_name: str) -> None:
         node = self.nodes[node_name]
@@ -425,6 +427,10 @@ class LangGraph:  # pylint: disable=too-many-instance-attributes
 
     def stop(self) -> None:
         self.is_running = False
+        for disposable in self._subscriptions:
+            with contextlib.suppress(Exception):
+                disposable.dispose()
+        self._subscriptions.clear()
         self._executor_pool.shutdown(wait=True, cancel_futures=True)
 
     def get_state(self) -> GraphState: