diff --git a/features/steps/tdd_mcp_client_start_race_steps.py b/features/steps/tdd_mcp_client_start_race_steps.py new file mode 100644 index 000000000..df42a1ce1 --- /dev/null +++ b/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}" + ) diff --git a/features/tdd_mcp_client_start_race.feature b/features/tdd_mcp_client_start_race.feature new file mode 100644 index 000000000..19adc2d94 --- /dev/null +++ b/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