Merge pull request 'Fix race condition in McpClient.start() double initialization' (#10892) from bugfix/mcp-race-condition-start into master

CHANGELOG.md

@@ -269,7 +269,12 @@ ensuring data is stored with proper parameter values.
   on a completed plan would silently destroy the ``cleveragents/plan-<id>`` git
   worktree branch, causing ``plan apply`` to merge zero artifacts.  The guard
   preserves the branch per spec (§sandbox.cleanup defaults to ``on_apply``).
-
+- **Race condition in ``McpClient.start()`` allows concurrent double initialisation** (#10438):
+  Added ``_state == McpClientState.STARTING`` check inside the ``threading.RLock`` in
+  ``start()`` so that concurrent callers see the in-progress state and return immediately,
+  preventing double initialisation of the MCP server connection, resource leaks, and state
+  corruption. TDD regression test added with BDD scenarios covering concurrent and sequential
+  start paths.
 - **Global CLI options ``--data-dir``, ``--config-path``, and ``-v`` now work correctly**
   (#6785): These spec-required flags were absent from ``main_callback()`` in
   ``src/cleveragents/cli/main.py``, causing any invocation with these flags to crash

CONTRIBUTORS.md

@@ -17,6 +17,9 @@
 
 * HAL9000 <HAL9000@cleverthis.com> has contributed CLI rendering improvements and TUI overlay visibility handling for `agents project context set` output.
 
+
+     * Jeffrey Phillips Freeman has contributed the McpClient.start() race condition fix (#10438): added _state == STARTING guard inside threading.RLock in start() and _ensure_started(), ensuring concurrent callers return immediately when initialisation is already in progress.
+
 # Details
 
 Below are some of the specific details of various contributions.
@@ -85,6 +88,7 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the agent task memory leak fix (#9044): replaced `list.remove` with `set.discard` as the done_callback for asyncio tasks in `Agent._tasks`, preventing unbounded memory growth in long-lived agents and ensuring safe concurrent task removal.
 * HAL 9000 has contributed the ACMS context show/clear CLI commands (PR #9675 / issue #9586): implemented `context show <view>` displaying assembled context with per-tier budget utilization summary (hot/warm/cold), and `context clear` with --path, --tag, --tier filtering plus confirmation prompt with --yes bypass. Includes 12 Behave BDD scenarios, 9 Robot Framework integration tests, ASV benchmarks, full type annotations, and _TierServiceProtocol for type safety.
 * HAL 9000 <hal9000@cleverthis.com> has contributed cost and session budget tracking and enforcement (issue #8609): implemented `CostTrackingService`, `BudgetExceededError`, three-tier budget hierarchy (plan/session/organization), warning thresholds at 90% utilization, per-provider cost tracking, and comprehensive BDD test suite.
+* HAL 9000 has contributed the AutoDebug node state mutation fix (#10496): fixed _analyze_error, _generate_fix, and _validate_fix in src/cleveragents/agents/graphs/auto_debug.py to return partial update dicts per LangGraph s node contract, preventing duplicate state entries and checkpoint inconsistencies.
 
 # Details (PR Contributions)
 

features/mocks/counting_mcp_transport.py

@@ -0,0 +1,49 @@
+"""Counting MCP transport for race-condition testing.
+
+This module provides :class:`CountingMCPTransport`, a test double that wraps a
+``MockMCPTransport`` and records how many times ``connect()`` and
+``discover_tools()`` (tools/list) are called.  It uses thread-safe counters so
+that concurrent scenarios can assert the exact call counts expected when fixing
+race conditions in ``McpClient.start()`` (issue #10438).
+
+See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from cleveragents.mcp.adapter import MCPServerConfig, MCPTransport
+from features.mocks.mock_mcp_transport import MockMCPTransport
+
+
+class CountingMCPTransport(MCPTransport):
+    """Transport that counts ``connect()`` and ``discover_tools()`` calls.
+
+    Wraps an inner transport (typically :class:`MockMCPTransport`) and
+    increments thread-safe counters for each ``connect()`` call and every
+    ``tools/list`` method call on ``call()``.  Used to verify that concurrent
+    :meth:`~McpClient.start` invocations produce exactly one initialisation
+    sequence.
+    """
+
+    def __init__(self, tools: list[dict[str, Any]]) -> None:
+        self._inner = MockMCPTransport(tools=tools)
+        self._lock: threading.Lock = threading.Lock()
+        self.connect_count: int = 0
+        self.discover_count: int = 0
+
+    def connect(self, config: MCPServerConfig) -> dict[str, Any]:
+        with self._lock:
+            self.connect_count += 1
+        return self._inner.connect(config)
+
+    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
+        if method == "tools/list":
+            with self._lock:
+                self.discover_count += 1
+        return self._inner.call(method, params)
+
+    def close(self) -> None:
+        self._inner.close()

features/steps/tdd_mcp_client_race_condition_start_steps.py

@@ -0,0 +1,187 @@
+"""Step definitions for TDD issue #10438 — McpClient.start() race condition.
+
+This module provides the Behave step implementations for the TDD bug-capture
+test that proves and verifies the fix for bug #10438.
+
+The bug was a race condition in ``McpClient.start()``: the method released the
+``threading.RLock`` after checking ``_started`` but before calling
+``connect()`` and ``discover_tools()``.  Concurrent callers could both pass
+the idempotency check and initialise the MCP server connection multiple times,
+causing resource leaks and state corruption.
+
+The fix adds ``_state == McpClientState.STARTING`` to the guard inside the
+lock so that a second concurrent caller returns immediately without calling
+``connect()`` or ``discover_tools()`` again.
+
+Race detection uses a counting transport that records how many times
+``connect()`` and ``discover_tools()`` are called.  A ``threading.Barrier``
+synchronises the threads so they all enter ``start()`` at approximately the
+same instant, maximising the chance of the race manifesting.
+
+See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
+"""
+
+from __future__ import annotations
+
+import threading
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.mcp.adapter import MCPServerConfig
+from cleveragents.mcp.client import McpClient, McpClientConfig, McpClientState
+from features.mocks.counting_mcp_transport import CountingMCPTransport
+
+
+# ── Given ─────────────────────────────────────────────────────────
+
+
+@given("an MCP client configured for concurrent start testing")
+def step_mcp_client_for_concurrent_start(context: Context) -> None:
+    """Create an MCP client with a counting transport for race testing.
+
+    The counting transport records how many times ``connect()`` and
+    ``discover_tools()`` are called so that the test can assert exactly
+    one invocation of each after concurrent ``start()`` calls.
+    """
+    tools = [{"name": "test_tool", "description": "Test tool", "inputSchema": {}}]
+    transport = CountingMCPTransport(tools=tools)
+
+    config = McpClientConfig(
+        server=MCPServerConfig(name="test-server", transport="stdio", command="echo"),
+        lazy_start=False,
+        idle_timeout_seconds=0,
+        health_check_interval_seconds=0,
+    )
+    client = McpClient(config=config, transport=transport)
+
+    context.mcp_race_client = client
+    context.mcp_race_transport = transport
+    context.mcp_race_thread_errors: list[Exception] = []
+    context.mcp_race_thread_count = 0
+
+
+# ── When ──────────────────────────────────────────────────────────
+
+
+@when("{n:d} threads call start() simultaneously through a barrier")
+def step_concurrent_start(context: Context, n: int) -> None:
+    """Launch *n* threads that all call ``start()`` simultaneously.
+
+    A ``threading.Barrier`` synchronises the threads so they all enter
+    ``start()`` at (approximately) the same instant, maximising the chance
+    of the race manifesting.
+
+    The counting transport records how many times ``connect()`` and
+    ``discover_tools()`` are called.
+    """
+    client: McpClient = context.mcp_race_client
+    barrier = threading.Barrier(n)
+    errors: list[Exception] = []
+    collect_lock = threading.Lock()
+
+    def worker() -> None:
+        """Thread worker: wait on barrier, then call start()."""
+        try:
+            barrier.wait(timeout=10)
+            client.start()
+        except Exception as exc:
+            with collect_lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=worker, daemon=True) for _ in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=30)
+
+    # Verify no threads are still alive after join timeout
+    for t in threads:
+        assert not t.is_alive(), (
+            f"Thread {t.name} is still alive after join timeout — "
+            f"possible deadlock in start()"
+        )
+
+    barrier_errors = [
+        err for err in errors if isinstance(err, threading.BrokenBarrierError)
+    ]
+    assert not barrier_errors, (
+        "Barrier synchronization failed while coordinating concurrent "
+        f"start() calls. Barrier errors: {barrier_errors}"
+    )
+
+    context.mcp_race_thread_errors = errors
+    context.mcp_race_thread_count = n
+
+
+@when("I call start() sequentially {n:d} times")
+def step_sequential_start(context: Context, n: int) -> None:
+    """Call ``start()`` sequentially *n* times on the same client."""
+    client: McpClient = context.mcp_race_client
+    for _ in range(n):
+        client.start()
+
+    context.mcp_race_thread_errors = []
+    context.mcp_race_thread_count = n
+
+
+# ── Then ──────────────────────────────────────────────────────────
+
+
+@then("connect should have been called exactly once")
+def step_connect_called_once(context: Context) -> None:
+    """Assert that ``connect()`` was called exactly once.
+
+    With the bug present (no STARTING state guard), multiple threads would
+    pass the ``_started`` check and each call ``connect()``, so the count
+    would be > 1.
+
+    With the fix applied, only the first thread calls ``connect()`` and
+    subsequent concurrent callers see ``_state == STARTING`` and return.
+    """
+    transport: CountingMCPTransport = context.mcp_race_transport
+    count = transport.connect_count
+    assert count == 1, (
+        f"connect() was called {count} time(s), expected exactly 1. "
+        f"This indicates the race condition in McpClient.start() is present — "
+        f"multiple threads bypassed the idempotency guard. "
+        f"Thread errors (if any): {context.mcp_race_thread_errors}"
+    )
+
+
+@then("discover_tools should have been called exactly once")
+def step_discover_tools_called_once(context: Context) -> None:
+    """Assert that ``discover_tools()`` was called exactly once.
+
+    With the bug present, multiple threads would call ``discover_tools()``
+    concurrently, so the count would be > 1.
+
+    With the fix applied, only the first thread calls ``discover_tools()``
+    and subsequent concurrent callers return early.
+    """
+    transport: CountingMCPTransport = context.mcp_race_transport
+    count = transport.discover_count
+    assert count == 1, (
+        f"discover_tools() was called {count} time(s), expected exactly 1. "
+        f"This indicates the race condition in McpClient.start() is present — "
+        f"multiple threads bypassed the idempotency guard. "
+        f"Thread errors (if any): {context.mcp_race_thread_errors}"
+    )
+
+
+@then("the MCP client state should be running after concurrent starts")
+def step_client_state_running_after_concurrent(context: Context) -> None:
+    """Assert the client is in RUNNING state after concurrent start() calls."""
+    client: McpClient = context.mcp_race_client
+    actual = client.state
+    assert actual == McpClientState.RUNNING, (
+        f"Expected client state to be '{McpClientState.RUNNING}', got '{actual}'. "
+        f"Thread errors (if any): {context.mcp_race_thread_errors}"
+    )
+
+
+@then("all concurrent start threads should have completed without error")
+def step_all_threads_no_error(context: Context) -> None:
+    """Assert that all concurrent start() threads completed without error."""
+    errors = context.mcp_race_thread_errors
+    assert not errors, f"Expected no thread errors but got {len(errors)}: {errors}"

features/tdd_mcp_client_race_condition_start.feature

@@ -0,0 +1,39 @@
+# This test captures bug #10438 — McpClient.start() race condition allowing
+# concurrent double initialisation.
+#
+# The fix adds a check for ``_state == McpClientState.STARTING`` inside the
+# lock so that a second concurrent caller sees the in-progress state and
+# returns immediately, preventing double initialisation.
+#
+# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
+
+@tdd_issue @tdd_issue_10438
+Feature: TDD Issue #10438 — McpClient.start() race condition allows concurrent double initialisation
+  McpClient.start() previously released the threading.RLock after checking
+  ``_started`` but before calling ``connect()`` and ``discover_tools()``.
+  Concurrent callers could both pass the idempotency check and initialise
+  the MCP server connection multiple times, causing resource leaks and state
+  corruption.
+
+  The fix adds ``_state == McpClientState.STARTING`` to the guard inside the
+  lock so that a second concurrent caller returns immediately.
+
+  Scenario: Concurrent start() calls result in exactly one connect() invocation
+    Given an MCP client configured for concurrent start testing
+    When 5 threads call start() simultaneously through a barrier
+    Then connect should have been called exactly once
+    And discover_tools should have been called exactly once
+    And the MCP client state should be running after concurrent starts
+
+  Scenario: Concurrent start() calls leave the client in RUNNING state
+    Given an MCP client configured for concurrent start testing
+    When 10 threads call start() simultaneously through a barrier
+    Then the MCP client state should be running after concurrent starts
+    And all concurrent start threads should have completed without error
+
+  Scenario: Sequential start() calls remain idempotent after the fix
+    Given an MCP client configured for concurrent start testing
+    When I call start() sequentially 3 times
+    Then connect should have been called exactly once
+    And discover_tools should have been called exactly once
+    And the MCP client state should be running after concurrent starts

features/tdd_mcp_client_start_race.feature

@@ -6,7 +6,7 @@
 #
 # See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
 
-@tdd_expected_fail @tdd_issue @tdd_issue_10438
+@tdd_issue @tdd_issue_10438
 Feature: TDD Issue #10438 — McpClient.start() race condition allows concurrent double initialization
   McpClient.start() releases the threading.RLock after setting _state to STARTING
   but before calling connect() and discover_tools().  If two threads call start()

src/cleveragents/mcp/client.py

@@ -207,7 +207,7 @@ class McpClient:
             If the MCP server cannot be reached.
         """
         with self._lock:
-            if self._started:
+            if self._started or self._state == McpClientState.STARTING:
                 return
             self._state = McpClientState.STARTING
 
@@ -263,7 +263,7 @@ class McpClient:
             If the client is not started and lazy_start is disabled.
         """
         with self._lock:
-            if self._started:
+            if self._started or self._state == McpClientState.STARTING:
                 return
             if not self._config.lazy_start:
                 msg = (