features/steps/tdd_mcp_client_start_race_steps.py

@@ -0,0 +1,237 @@
+"""Step definitions for features/tdd_mcp_client_start_race.feature.
+
+TDD issue-capture scenario for #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().  Two concurrent
+callers can both pass the _started idempotency check (still False) and both
+proceed to call connect() and discover_tools() simultaneously.
+
+The test uses @tdd_expected_fail so that CI passes while the bug is unfixed.
+The underlying assertion ("connect() called exactly once") fails because the
+race allows multiple threads to trigger initialization.  The tag inversion
+causes the scenario to be reported as passed.
+
+Race detection uses a threading.Barrier to synchronise threads so they all
+enter start() at approximately the same instant, maximising the chance of
+the race manifesting.  A counting mock transport records how many times
+connect() and discover_tools() are called.
+
+When bug #10438 is fixed (by checking _state == STARTING inside the lock),
+the assertions will pass and the @tdd_expected_fail tag must be removed.
+
+See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.mcp.adapter import MCPServerConfig
+from cleveragents.mcp.client import McpClient, McpClientConfig
+from features.mocks.mock_mcp_transport import MockMCPTransport
+
+
+class _CountingMockTransport(MockMCPTransport):
+    """Mock transport that counts connect() and discover_tools() calls.
+
+    Uses a threading.Lock to make the counters thread-safe so that
+    concurrent calls are accurately recorded.
+    """
+
+    def __init__(self) -> None:
+        tools = [
+            {
+                "name": "test_tool",
+                "description": "Mock tool for race condition test",
+                "inputSchema": {},
+            }
+        ]
+        super().__init__(tools=tools)
+        self._counter_lock = threading.Lock()
+        self.connect_call_count: int = 0
+        self.discover_call_count: int = 0
+        # Add a small sleep inside connect() to widen the race window,
+        # making it more likely that multiple threads enter concurrently.
+        self._connect_delay: float = 0.02
+
+    def connect(self, config: MCPServerConfig) -> dict[str, Any]:
+        with self._counter_lock:
+            self.connect_call_count += 1
+        # Sleep outside the counter lock to widen the race window.
+        # This gives other threads time to also pass the _started check
+        # and enter connect() before the first thread finishes.
+        time.sleep(self._connect_delay)
+        return super().connect(config)
+
+    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
+        if method == "tools/list":
+            with self._counter_lock:
+                self.discover_call_count += 1
+        return super().call(method, params)
+
+
+# ── Given ─────────────────────────────────────────────────────────
+
+
+@given("an McpClient with a counting mock transport")
+def step_mcp_client_counting_transport(context: Context) -> None:
+    """Create an McpClient backed by a counting mock transport.
+
+    The client is configured with lazy_start=False, idle_timeout=0, and
+    health_check_interval=0 so that only explicit start() calls trigger
+    connect() and discover_tools().
+    """
+    context.counting_transport = _CountingMockTransport()
+    server_config = MCPServerConfig(
+        name="test-server",
+        transport="stdio",
+        command="echo",
+    )
+    client_config = McpClientConfig(
+        server=server_config,
+        lazy_start=False,
+        idle_timeout_seconds=0,
+        health_check_interval_seconds=0,
+    )
+    context.race_client = McpClient(
+        config=client_config,
+        transport=context.counting_transport,
+    )
+    context.race_thread_errors: list[Exception] = []
+    context.race_threads_completed: int = 0
+
+
+# ── When ──────────────────────────────────────────────────────────
+
+
+@when("{n:d} threads call start() concurrently 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 condition manifesting.
+
+    With the bug present, multiple threads will pass the _started check
+    (still False) and each call connect() and discover_tools().
+    With the fix applied, only one thread proceeds past the STARTING
+    state check and the others return early.
+    """
+    barrier = threading.Barrier(n)
+    errors: list[Exception] = []
+    completed: list[int] = [0]
+    collect_lock = threading.Lock()
+
+    def worker() -> None:
+        """Thread worker: wait on barrier, then call start()."""
+        try:
+            barrier.wait(timeout=10)
+            context.race_client.start()
+            with collect_lock:
+                completed[0] += 1
+        except Exception as exc:
+            with collect_lock:
+                errors.append(exc)
+                completed[0] += 1
+
+    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 McpClient.start()"
+        )
+
+    barrier_errors = [
+        err for err in errors if isinstance(err, threading.BrokenBarrierError)
+    ]
+    assert not barrier_errors, (
+        "Barrier synchronization failed while coordinating concurrent "
+        "start() calls. This invalidates race setup. "
+        f"Barrier errors: {barrier_errors}; all errors: {errors}"
+    )
+
+    context.race_thread_errors = errors
+    context.race_threads_completed = completed[0]
+    context.race_thread_count = n
+
+
+# ── Then ──────────────────────────────────────────────────────────
+
+
+@then("connect() should have been called exactly once")
+def step_connect_called_once(context: Context) -> None:
+    """Assert connect() was called exactly once.
+
+    With the bug present (no STARTING state check inside the lock),
+    multiple threads will enter the start() body and each call connect(),
+    so the counter will be > 1.
+
+    This assertion therefore FAILS while the bug exists — which is the
+    expected behavior for a @tdd_expected_fail test.
+
+    When the fix adds the _state == STARTING check inside the lock, only
+    the first thread calls connect() and the count is exactly 1.
+    """
+    count = context.counting_transport.connect_call_count
+    assert count == 1, (
+        f"connect() was called {count} time(s), expected exactly 1. "
+        f"This confirms the race condition in McpClient.start() — multiple "
+        f"threads passed the _started check because the STARTING state is "
+        f"not checked inside the lock before proceeding. "
+        f"Thread errors (if any): {context.race_thread_errors}"
+    )
+
+
+@then("discover_tools() should have been called exactly once")
+def step_discover_tools_called_once(context: Context) -> None:
+    """Assert discover_tools() was called exactly once.
+
+    With the bug present, multiple threads call discover_tools() concurrently.
+    With the fix, only one thread proceeds to call discover_tools().
+    """
+    count = context.counting_transport.discover_call_count
+    assert count == 1, (
+        f"discover_tools() was called {count} time(s), expected exactly 1. "
+        f"This confirms the race condition in McpClient.start() — multiple "
+        f"threads proceeded past the idempotency check. "
+        f"Thread errors (if any): {context.race_thread_errors}"
+    )
+
+
+@then('the client state should be "{expected_state}"')
+def step_client_state_is(context: Context, expected_state: str) -> None:
+    """Assert the client is in the expected state after concurrent starts."""
+    actual = context.race_client.state
+    assert actual == expected_state, (
+        f"Expected client state '{expected_state}', got '{actual}'. "
+        f"Thread errors (if any): {context.race_thread_errors}"
+    )
+
+
+@then("all {n:d} threads should have completed without hanging")
+def step_all_threads_completed_no_hang(context: Context, n: int) -> None:
+    """Verify that all *n* threads completed (either success or error).
+
+    The total of completed threads must equal the number launched.
+    This confirms no threads silently hung or deadlocked.
+    """
+    total = context.race_threads_completed
+    assert total == n, (
+        f"Expected {n} threads to complete but only {total} did. "
+        "A mismatch indicates a hung thread or possible deadlock in "
+        "McpClient.start(). "
+        f"Thread errors (if any): {context.race_thread_errors}"
+    )

features/tdd_mcp_client_start_race.feature

@@ -0,0 +1,27 @@
+# This test captures bug #10438 — McpClient.start() race condition.
+#
+# The @tdd_expected_fail tag inverts the test result: the underlying assertion
+# fails (proving the bug exists) but CI reports the scenario as passed.
+# When #10438 is fixed, the @tdd_expected_fail tag must be removed.
+#
+# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
+
+@tdd_expected_fail @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()
+  concurrently, both can pass the _started idempotency check (which is still False)
+  and both proceed to call connect() and discover_tools() simultaneously.
+
+  This test proves the race exists by launching multiple threads that call start()
+  simultaneously via a threading.Barrier and verifying that connect() is called
+  exactly once (the correct, thread-safe behavior).  Without the bug present,
+  connect() is called exactly once.  With the bug, connect() is called multiple times.
+
+  Scenario: Concurrent start() calls must invoke connect() exactly once
+    Given an McpClient with a counting mock transport
+    When 5 threads call start() concurrently through a barrier
+    Then connect() should have been called exactly once
+    And discover_tools() should have been called exactly once
+    And the client state should be "running"
+    And all 5 threads should have completed without hanging