From 908e53ea8522b89a149e4fec584b9a6fd31e60b6 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 19 Apr 2026 12:54:07 +0000 Subject: [PATCH 1/8] fix(mcp): release RLock before transport call in MCPToolAdapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCPToolAdapter.invoke() and discover_tools() held the RLock during the entire transport.call(), blocking all concurrent operations on the adapter. The fix splits both methods into three phases: 1. Acquire lock — validate connection state and inputs 2. Release lock — make the transport call without holding the lock 3. Re-acquire lock (discover_tools only) — update shared state Added Behave BDD tests verifying: - Concurrent invocations complete in parallel (wall-clock < serial time) - Concurrent discoveries complete in parallel - Lock is provably not held during transport calls (cross-thread check) - Validation still occurs under lock before transport call - Disconnected adapter still raises RuntimeError under lock ISSUES CLOSED: #10512 --- ...tdd_mcp_adapter_rlock_concurrency_steps.py | 289 ++++++++++++++++++ .../tdd_mcp_adapter_rlock_concurrency.feature | 52 ++++ src/cleveragents/mcp/adapter.py | 147 +++++---- 3 files changed, 426 insertions(+), 62 deletions(-) create mode 100644 features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py create mode 100644 features/tdd_mcp_adapter_rlock_concurrency.feature diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py new file mode 100644 index 000000000..5f5b5cfa6 --- /dev/null +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -0,0 +1,289 @@ +"""Step definitions for features/tdd_mcp_adapter_rlock_concurrency.feature. + +Tests that MCPToolAdapter releases its RLock before making transport calls, +allowing concurrent operations to proceed without being blocked. + +Bug #10512: MCPToolAdapter holds RLock during entire transport call, +blocking concurrent operations. +""" + +from __future__ import annotations + +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.mcp.adapter import ( + MCPServerConfig, + MCPToolAdapter, + MCPToolResult, +) +from features.mocks.mock_mcp_transport import MockMCPTransport + + +def _mock_tool(name: str, desc: str = "", schema: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "name": name, + "description": desc or f"Mock tool {name}", + "inputSchema": schema or {}, + } + + +class SlowMCPTransport(MockMCPTransport): + """Transport that introduces a configurable delay during call().""" + + def __init__( + self, + delay: float, + tools: list[dict[str, Any]] | None = None, + ) -> None: + super().__init__(tools=tools) + self._delay = delay + + def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + # Use the real time.sleep (not the capped test version) to ensure + # the delay is meaningful for concurrency testing. + time._original_sleep(self._delay) # type: ignore[attr-defined] + return super().call(method, params) + + +class LockCheckingInvokeTransport(MockMCPTransport): + """Transport that checks whether the adapter's RLock is held during call().""" + + def __init__( + self, + adapter_lock: threading.RLock, + tools: list[dict[str, Any]] | None = None, + ) -> None: + super().__init__(tools=tools) + self._adapter_lock = adapter_lock + self.lock_was_held_during_call = False + + def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + if method == "tools/call": + # Try to acquire the lock from a different thread. + # If the lock is held by the calling thread, a different thread + # will fail to acquire it (non-blocking). + acquired_event = threading.Event() + lock_held = [True] + + def _try_acquire() -> None: + got_it = self._adapter_lock.acquire(blocking=False) + if got_it: + lock_held[0] = False + self._adapter_lock.release() + acquired_event.set() + + t = threading.Thread(target=_try_acquire, daemon=True) + t.start() + t.join(timeout=1.0) + acquired_event.wait(timeout=1.0) + self.lock_was_held_during_call = lock_held[0] + + return super().call(method, params) + + +class LockCheckingDiscoveryTransport(MockMCPTransport): + """Transport that checks whether the adapter's RLock is held during tools/list.""" + + def __init__( + self, + adapter_lock: threading.RLock, + tools: list[dict[str, Any]] | None = None, + ) -> None: + super().__init__(tools=tools) + self._adapter_lock = adapter_lock + self.lock_was_held_during_call = False + + def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + if method == "tools/list": + acquired_event = threading.Event() + lock_held = [True] + + def _try_acquire() -> None: + got_it = self._adapter_lock.acquire(blocking=False) + if got_it: + lock_held[0] = False + self._adapter_lock.release() + acquired_event.set() + + t = threading.Thread(target=_try_acquire, daemon=True) + t.start() + t.join(timeout=1.0) + acquired_event.wait(timeout=1.0) + self.lock_was_held_during_call = lock_held[0] + + return super().call(method, params) + + +# --------------------------------------------------------------- +# Given Steps +# --------------------------------------------------------------- + + +@given('a connected MCP adapter with a slow transport tool "{tool_name}" taking {delay:f} seconds') +def step_slow_transport_tool(context: Context, tool_name: str, delay: float) -> None: + tools = [_mock_tool(tool_name)] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + context.mcp_transport = SlowMCPTransport(delay=delay, tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=context.mcp_transport) + context.mcp_adapter.connect() + context.mcp_adapter.discover_tools() + + +@given('a connected MCP adapter with a slow discovery transport taking {delay:f} seconds') +def step_slow_discovery_transport(context: Context, delay: float) -> None: + tools = [_mock_tool("tool_0"), _mock_tool("tool_1")] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + context.mcp_transport = SlowMCPTransport(delay=delay, tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=context.mcp_transport) + context.mcp_adapter.connect() + + +@given('a connected MCP adapter with a lock-checking transport tool "{tool_name}"') +def step_lock_checking_invoke_transport(context: Context, tool_name: str) -> None: + tools = [_mock_tool(tool_name)] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + # Create adapter first to get its lock, then set up transport + transport_placeholder = MockMCPTransport(tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=transport_placeholder) + # Now create the lock-checking transport with the adapter's actual lock + lock_transport = LockCheckingInvokeTransport( + adapter_lock=context.mcp_adapter._lock, + tools=tools, + ) + # Replace the transport + context.mcp_adapter._transport = lock_transport + context.mcp_transport = lock_transport + context.mcp_adapter._connected = True + context.mcp_adapter._capabilities = {"tools": True} + context.mcp_adapter.discover_tools() + + +@given('a connected MCP adapter with a lock-checking discovery transport') +def step_lock_checking_discovery_transport(context: Context) -> None: + tools = [_mock_tool("tool_0"), _mock_tool("tool_1")] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + transport_placeholder = MockMCPTransport(tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=transport_placeholder) + lock_transport = LockCheckingDiscoveryTransport( + adapter_lock=context.mcp_adapter._lock, + tools=tools, + ) + context.mcp_adapter._transport = lock_transport + context.mcp_transport = lock_transport + context.mcp_adapter._connected = True + context.mcp_adapter._capabilities = {"tools": True} + + +# --------------------------------------------------------------- +# When Steps +# --------------------------------------------------------------- + + +@when('I invoke "{tool_name}" concurrently from {count:d} threads') +def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None: + results: list[MCPToolResult] = [] + start = time.monotonic() + + with ThreadPoolExecutor(max_workers=count) as executor: + futures = [ + executor.submit(context.mcp_adapter.invoke, tool_name, {}) + for _ in range(count) + ] + for future in as_completed(futures): + results.append(future.result()) + + context.concurrent_invoke_results = results + context.concurrent_wall_clock = time.monotonic() - start + + +@when('I call discover_tools concurrently from {count:d} threads') +def step_concurrent_discover(context: Context, count: int) -> None: + results: list[Any] = [] + errors: list[Exception] = [] + start = time.monotonic() + + with ThreadPoolExecutor(max_workers=count) as executor: + futures = [ + executor.submit(context.mcp_adapter.discover_tools) + for _ in range(count) + ] + for future in as_completed(futures): + try: + results.append(future.result()) + except Exception as exc: + errors.append(exc) + + context.concurrent_discover_results = results + context.concurrent_discover_errors = errors + context.concurrent_discover_wall_clock = time.monotonic() - start + + +# --------------------------------------------------------------- +# Then Steps +# --------------------------------------------------------------- + + +@then('all {count:d} concurrent invocations should succeed') +def step_all_invocations_succeed(context: Context, count: int) -> None: + results = context.concurrent_invoke_results + assert len(results) == count, ( + f"Expected {count} results, got {len(results)}" + ) + for i, result in enumerate(results): + assert result.success, ( + f"Invocation {i} failed: {result.error}" + ) + + +@then('the total wall-clock time should be less than {max_seconds:f} seconds') +def step_wall_clock_less_than(context: Context, max_seconds: float) -> None: + actual = context.concurrent_wall_clock + assert actual < max_seconds, ( + f"Total wall-clock time {actual:.3f}s >= {max_seconds}s. " + f"Concurrent calls appear to be serialized (lock held during transport call)." + ) + + +@then('all {count:d} concurrent discoveries should succeed') +def step_all_discoveries_succeed(context: Context, count: int) -> None: + results = context.concurrent_discover_results + errors = context.concurrent_discover_errors + assert not errors, ( + f"Expected no errors, got {len(errors)}: {errors}" + ) + assert len(results) == count, ( + f"Expected {count} results, got {len(results)}" + ) + + +@then('the total discovery wall-clock time should be less than {max_seconds:f} seconds') +def step_discovery_wall_clock_less_than(context: Context, max_seconds: float) -> None: + actual = context.concurrent_discover_wall_clock + assert actual < max_seconds, ( + f"Total discovery wall-clock time {actual:.3f}s >= {max_seconds}s. " + f"Concurrent calls appear to be serialized (lock held during transport call)." + ) + + +@then('the lock should not have been held during the transport call') +def step_lock_not_held_during_invoke(context: Context) -> None: + transport = context.mcp_transport + assert not transport.lock_was_held_during_call, ( + "The adapter's RLock was held during the transport.call() invocation. " + "The lock should be released before making the transport call." + ) + + +@then('the lock should not have been held during the discovery transport call') +def step_lock_not_held_during_discovery(context: Context) -> None: + transport = context.mcp_transport + assert not transport.lock_was_held_during_call, ( + "The adapter's RLock was held during the transport.call('tools/list') invocation. " + "The lock should be released before making the transport call." + ) diff --git a/features/tdd_mcp_adapter_rlock_concurrency.feature b/features/tdd_mcp_adapter_rlock_concurrency.feature new file mode 100644 index 000000000..34c1597ae --- /dev/null +++ b/features/tdd_mcp_adapter_rlock_concurrency.feature @@ -0,0 +1,52 @@ +@tdd_issue @tdd_issue_10512 +Feature: MCPToolAdapter releases RLock during transport calls + As a developer using MCPToolAdapter concurrently + I want the adapter to release its RLock before making transport calls + So that concurrent operations are not blocked by slow transport calls + + # ------------------------------------------------------------------- + # Bug #10512: MCPToolAdapter holds RLock during entire transport call, + # blocking concurrent operations. + # + # The fix releases the lock before the transport.call() and + # re-acquires it only when shared state must be mutated. + # ------------------------------------------------------------------- + + Scenario: Concurrent invoke calls are not blocked by a slow transport + Given a connected MCP adapter with a slow transport tool "slow_op" taking 0.3 seconds + When I invoke "slow_op" concurrently from 3 threads + Then all 3 concurrent invocations should succeed + And the total wall-clock time should be less than 0.9 seconds + + Scenario: Concurrent discover_tools calls are not blocked by a slow transport + Given a connected MCP adapter with a slow discovery transport taking 0.3 seconds + When I call discover_tools concurrently from 3 threads + Then all 3 concurrent discoveries should succeed + And the total discovery wall-clock time should be less than 0.9 seconds + + Scenario: Lock is not held during invoke transport call + Given a connected MCP adapter with a lock-checking transport tool "check_lock" + When I invoke "check_lock" with arguments {} + Then the invocation should succeed + And the lock should not have been held during the transport call + + Scenario: Lock is not held during discover_tools transport call + Given a connected MCP adapter with a lock-checking discovery transport + When I discover tools from the adapter + Then the lock should not have been held during the discovery transport call + + Scenario: Invoke still validates under lock before transport call + Given a connected MCP adapter with a slow transport tool "slow_op" taking 0.1 seconds + When I invoke "nonexistent_tool" with arguments {} + Then the invocation should fail + And the invocation error should mention "not found" + + Scenario: Invoke on disconnected adapter still raises under lock + Given an MCP adapter with a mock transport + When I invoke MCP tool "any_tool" while disconnected + Then the adapter error should mention "not connected" + + Scenario: Discover on disconnected adapter still raises under lock + Given an MCP adapter with a mock transport + When I discover tools expecting an error + Then the adapter error should mention "not connected" diff --git a/src/cleveragents/mcp/adapter.py b/src/cleveragents/mcp/adapter.py index 7be616705..7beda1e14 100644 --- a/src/cleveragents/mcp/adapter.py +++ b/src/cleveragents/mcp/adapter.py @@ -174,6 +174,12 @@ class MCPTransport: class MCPToolAdapter: """Adapter bridging an MCP server into the CleverAgents ToolRegistry. + Thread-safety strategy: the adapter's ``RLock`` protects shared + mutable state (``_connected``, ``_capabilities``, ``_tools``, + ``_notification_listeners``). The lock is **released** before + making transport calls (``_transport.call()``) so that concurrent + operations are not blocked by slow network I/O. See bug #10512. + Parameters ---------- config: @@ -411,6 +417,10 @@ class MCPToolAdapter: ) -> list[MCPToolDescriptor]: """Enumerate tools from the connected MCP server. + The adapter's RLock is held only for state validation and + mutation — it is released before the transport call so that + concurrent operations are not blocked by slow network I/O. + Parameters ---------- tool_filter: @@ -421,6 +431,7 @@ class MCPToolAdapter: RuntimeError If the adapter is not connected. """ + # Phase 1: Validate connection state under lock. with self._lock: if not self._connected: msg = ( @@ -428,25 +439,30 @@ class MCPToolAdapter: f"Call connect() first." ) raise RuntimeError(msg) + transport = self._transport - result = self._transport.call("tools/list", {}) - raw_tools = result.get("tools", []) + # Phase 2: Transport call WITHOUT lock — allows concurrent access. + result = transport.call("tools/list", {}) + raw_tools = result.get("tools", []) - descriptors: list[MCPToolDescriptor] = [] - for raw in raw_tools: - desc = MCPToolDescriptor( - name=raw.get("name", ""), - description=raw.get("description", ""), - input_schema=raw.get("inputSchema", {}), - annotations=raw.get("annotations", {}), - ) - descriptors.append(desc) + descriptors: list[MCPToolDescriptor] = [] + for raw in raw_tools: + desc = MCPToolDescriptor( + name=raw.get("name", ""), + description=raw.get("description", ""), + input_schema=raw.get("inputSchema", {}), + annotations=raw.get("annotations", {}), + ) + descriptors.append(desc) - if tool_filter: - descriptors = self._apply_filter(descriptors, tool_filter) + if tool_filter: + descriptors = self._apply_filter(descriptors, tool_filter) + # Phase 3: Update shared state under lock. + with self._lock: self._tools = {d.name: d for d in descriptors} - return descriptors + + return descriptors def invoke( self, @@ -456,7 +472,9 @@ class MCPToolAdapter: """Invoke a discovered MCP tool by name. Validates inputs against the tool's JSON Schema before calling - the server. + the server. The adapter's RLock is held only for state + validation — it is released before the transport call so that + concurrent invocations are not blocked by slow network I/O. Parameters ---------- @@ -470,6 +488,7 @@ class MCPToolAdapter: MCPToolResult Invocation outcome with success flag, data, and timing. """ + # Phase 1: Validate state and inputs under lock. with self._lock: if not self._connected: msg = ( @@ -494,57 +513,61 @@ class MCPToolAdapter: f"{validation_error}", ) - start = time.monotonic() - try: - result = self._transport.call( - "tools/call", - {"name": tool_name, "arguments": arguments}, - ) - except TimeoutError as exc: - elapsed = (time.monotonic() - start) * 1000 - return MCPToolResult( - success=False, - error=f"Tool '{tool_name}' timeout: {exc}", - duration_ms=elapsed, - ) - except Exception as exc: - elapsed = (time.monotonic() - start) * 1000 - return MCPToolResult( - success=False, - error=f"MCP server error invoking '{tool_name}': {exc}", - duration_ms=elapsed, - ) + transport = self._transport + + # Phase 2: Transport call WITHOUT lock — allows concurrent access. + start = time.monotonic() + try: + result = transport.call( + "tools/call", + {"name": tool_name, "arguments": arguments}, + ) + except TimeoutError as exc: elapsed = (time.monotonic() - start) * 1000 - - if result.get("isError"): - content = result.get("content", []) - if content and isinstance(content, list) and len(content) > 0: - error_text = content[0].get("text", "unknown error") - else: - error_text = "unknown error" - return MCPToolResult( - success=False, - error=f"MCP server error: {error_text}", - duration_ms=elapsed, - ) - - # MCP 1.4.0 returns ``content`` as a list of ContentItem dicts - # (e.g. ``[{"type": "text", "text": "..."}]``). Normalise to a - # plain dict so that ``MCPToolResult.data`` is always - # ``dict[str, Any]`` and downstream code can rely on dict access. - raw_content = result.get("content", []) - if isinstance(raw_content, list): - normalised: dict[str, Any] = {"content": raw_content} - elif isinstance(raw_content, dict): - normalised = raw_content - else: - # Unexpected type — wrap it so the dict contract holds. - normalised = {"content": raw_content} return MCPToolResult( - success=True, - data=normalised, + success=False, + error=f"Tool '{tool_name}' timeout: {exc}", duration_ms=elapsed, ) + except Exception as exc: + elapsed = (time.monotonic() - start) * 1000 + return MCPToolResult( + success=False, + error=f"MCP server error invoking '{tool_name}': {exc}", + duration_ms=elapsed, + ) + elapsed = (time.monotonic() - start) * 1000 + + # Phase 3: Process result (no lock needed — local variables only). + if result.get("isError"): + content = result.get("content", []) + if content and isinstance(content, list) and len(content) > 0: + error_text = content[0].get("text", "unknown error") + else: + error_text = "unknown error" + return MCPToolResult( + success=False, + error=f"MCP server error: {error_text}", + duration_ms=elapsed, + ) + + # MCP 1.4.0 returns ``content`` as a list of ContentItem dicts + # (e.g. ``[{"type": "text", "text": "..."}]``). Normalise to a + # plain dict so that ``MCPToolResult.data`` is always + # ``dict[str, Any]`` and downstream code can rely on dict access. + raw_content = result.get("content", []) + if isinstance(raw_content, list): + normalised: dict[str, Any] = {"content": raw_content} + elif isinstance(raw_content, dict): + normalised = raw_content + else: + # Unexpected type — wrap it so the dict contract holds. + normalised = {"content": raw_content} + return MCPToolResult( + success=True, + data=normalised, + duration_ms=elapsed, + ) def register_tools( self, -- 2.52.0 From 69b7cd3cb219e864e047f754a261b32b3cbf43a8 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Wed, 22 Apr 2026 06:00:47 +0000 Subject: [PATCH 2/8] fix(mcp): release RLock before transport call in MCPToolAdapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCPToolAdapter.invoke() and discover_tools() held the RLock during the entire transport.call(), blocking all concurrent operations on the adapter. The fix splits both methods into three phases: 1. Acquire lock — validate connection state and inputs 2. Release lock — make the transport call without holding the lock 3. Re-acquire lock (discover_tools only) — update shared state Added Behave BDD tests verifying: - Concurrent invocations complete in parallel (wall-clock < serial time) - Concurrent discoveries complete in parallel - Lock is provably not held during transport calls (cross-thread check) - Validation still occurs under lock before transport call - Disconnected adapter still raises RuntimeError under lock ISSUES CLOSED: #10512 --- ...tdd_mcp_adapter_rlock_concurrency_steps.py | 47 +++++++++---------- 1 file changed, 22 insertions(+), 25 deletions(-) diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py index 5f5b5cfa6..b3198763b 100644 --- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -25,7 +25,9 @@ from cleveragents.mcp.adapter import ( from features.mocks.mock_mcp_transport import MockMCPTransport -def _mock_tool(name: str, desc: str = "", schema: dict[str, Any] | None = None) -> dict[str, Any]: +def _mock_tool( + name: str, desc: str = "", schema: dict[str, Any] | None = None +) -> dict[str, Any]: return { "name": name, "description": desc or f"Mock tool {name}", @@ -125,7 +127,9 @@ class LockCheckingDiscoveryTransport(MockMCPTransport): # --------------------------------------------------------------- -@given('a connected MCP adapter with a slow transport tool "{tool_name}" taking {delay:f} seconds') +@given( + 'a connected MCP adapter with a slow transport tool "{tool_name}" taking {delay:f} seconds' +) def step_slow_transport_tool(context: Context, tool_name: str, delay: float) -> None: tools = [_mock_tool(tool_name)] config = MCPServerConfig(name="test-server", transport="stdio", command="echo") @@ -135,7 +139,9 @@ def step_slow_transport_tool(context: Context, tool_name: str, delay: float) -> context.mcp_adapter.discover_tools() -@given('a connected MCP adapter with a slow discovery transport taking {delay:f} seconds') +@given( + "a connected MCP adapter with a slow discovery transport taking {delay:f} seconds" +) def step_slow_discovery_transport(context: Context, delay: float) -> None: tools = [_mock_tool("tool_0"), _mock_tool("tool_1")] config = MCPServerConfig(name="test-server", transport="stdio", command="echo") @@ -164,7 +170,7 @@ def step_lock_checking_invoke_transport(context: Context, tool_name: str) -> Non context.mcp_adapter.discover_tools() -@given('a connected MCP adapter with a lock-checking discovery transport') +@given("a connected MCP adapter with a lock-checking discovery transport") def step_lock_checking_discovery_transport(context: Context) -> None: tools = [_mock_tool("tool_0"), _mock_tool("tool_1")] config = MCPServerConfig(name="test-server", transport="stdio", command="echo") @@ -202,7 +208,7 @@ def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None context.concurrent_wall_clock = time.monotonic() - start -@when('I call discover_tools concurrently from {count:d} threads') +@when("I call discover_tools concurrently from {count:d} threads") def step_concurrent_discover(context: Context, count: int) -> None: results: list[Any] = [] errors: list[Exception] = [] @@ -210,8 +216,7 @@ def step_concurrent_discover(context: Context, count: int) -> None: with ThreadPoolExecutor(max_workers=count) as executor: futures = [ - executor.submit(context.mcp_adapter.discover_tools) - for _ in range(count) + executor.submit(context.mcp_adapter.discover_tools) for _ in range(count) ] for future in as_completed(futures): try: @@ -229,19 +234,15 @@ def step_concurrent_discover(context: Context, count: int) -> None: # --------------------------------------------------------------- -@then('all {count:d} concurrent invocations should succeed') +@then("all {count:d} concurrent invocations should succeed") def step_all_invocations_succeed(context: Context, count: int) -> None: results = context.concurrent_invoke_results - assert len(results) == count, ( - f"Expected {count} results, got {len(results)}" - ) + assert len(results) == count, f"Expected {count} results, got {len(results)}" for i, result in enumerate(results): - assert result.success, ( - f"Invocation {i} failed: {result.error}" - ) + assert result.success, f"Invocation {i} failed: {result.error}" -@then('the total wall-clock time should be less than {max_seconds:f} seconds') +@then("the total wall-clock time should be less than {max_seconds:f} seconds") def step_wall_clock_less_than(context: Context, max_seconds: float) -> None: actual = context.concurrent_wall_clock assert actual < max_seconds, ( @@ -250,19 +251,15 @@ def step_wall_clock_less_than(context: Context, max_seconds: float) -> None: ) -@then('all {count:d} concurrent discoveries should succeed') +@then("all {count:d} concurrent discoveries should succeed") def step_all_discoveries_succeed(context: Context, count: int) -> None: results = context.concurrent_discover_results errors = context.concurrent_discover_errors - assert not errors, ( - f"Expected no errors, got {len(errors)}: {errors}" - ) - assert len(results) == count, ( - f"Expected {count} results, got {len(results)}" - ) + assert not errors, f"Expected no errors, got {len(errors)}: {errors}" + assert len(results) == count, f"Expected {count} results, got {len(results)}" -@then('the total discovery wall-clock time should be less than {max_seconds:f} seconds') +@then("the total discovery wall-clock time should be less than {max_seconds:f} seconds") def step_discovery_wall_clock_less_than(context: Context, max_seconds: float) -> None: actual = context.concurrent_discover_wall_clock assert actual < max_seconds, ( @@ -271,7 +268,7 @@ def step_discovery_wall_clock_less_than(context: Context, max_seconds: float) -> ) -@then('the lock should not have been held during the transport call') +@then("the lock should not have been held during the transport call") def step_lock_not_held_during_invoke(context: Context) -> None: transport = context.mcp_transport assert not transport.lock_was_held_during_call, ( @@ -280,7 +277,7 @@ def step_lock_not_held_during_invoke(context: Context) -> None: ) -@then('the lock should not have been held during the discovery transport call') +@then("the lock should not have been held during the discovery transport call") def step_lock_not_held_during_discovery(context: Context) -> None: transport = context.mcp_transport assert not transport.lock_was_held_during_call, ( -- 2.52.0 From 79ecaea18f0ba65f7766a42114ea1f00f39d674b Mon Sep 17 00:00:00 2001 From: CleverThis Date: Wed, 22 Apr 2026 09:45:30 +0000 Subject: [PATCH 3/8] fix(tests): use getattr for time._original_sleep fallback in RLock concurrency tests --- features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py index b3198763b..0c94b72f7 100644 --- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -49,7 +49,8 @@ class SlowMCPTransport(MockMCPTransport): def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: # Use the real time.sleep (not the capped test version) to ensure # the delay is meaningful for concurrency testing. - time._original_sleep(self._delay) # type: ignore[attr-defined] + _real_sleep = getattr(time, '_original_sleep', time.sleep) + _real_sleep(self._delay) return super().call(method, params) -- 2.52.0 From 60671306ea3f7d853b35e8e84f86e429cd487265 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Wed, 22 Apr 2026 22:35:26 +0000 Subject: [PATCH 4/8] fix(tests): fix format and reduce timing thresholds in RLock concurrency tests Fixed ruff format violation (single quotes to double quotes) in step definitions file. Reduced SlowMCPTransport delay from 0.3s to 0.15s and tightened the wall-clock assertion from 0.9s to 0.40s to improve CI reliability while still proving concurrent execution. ISSUES CLOSED: #10512 --- features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py | 2 +- features/tdd_mcp_adapter_rlock_concurrency.feature | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py index 0c94b72f7..d574b61fb 100644 --- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -49,7 +49,7 @@ class SlowMCPTransport(MockMCPTransport): def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: # Use the real time.sleep (not the capped test version) to ensure # the delay is meaningful for concurrency testing. - _real_sleep = getattr(time, '_original_sleep', time.sleep) + _real_sleep = getattr(time, "_original_sleep", time.sleep) _real_sleep(self._delay) return super().call(method, params) diff --git a/features/tdd_mcp_adapter_rlock_concurrency.feature b/features/tdd_mcp_adapter_rlock_concurrency.feature index 34c1597ae..8129b0ddc 100644 --- a/features/tdd_mcp_adapter_rlock_concurrency.feature +++ b/features/tdd_mcp_adapter_rlock_concurrency.feature @@ -13,16 +13,16 @@ Feature: MCPToolAdapter releases RLock during transport calls # ------------------------------------------------------------------- Scenario: Concurrent invoke calls are not blocked by a slow transport - Given a connected MCP adapter with a slow transport tool "slow_op" taking 0.3 seconds + Given a connected MCP adapter with a slow transport tool "slow_op" taking 0.15 seconds When I invoke "slow_op" concurrently from 3 threads Then all 3 concurrent invocations should succeed - And the total wall-clock time should be less than 0.9 seconds + And the total wall-clock time should be less than 0.40 seconds Scenario: Concurrent discover_tools calls are not blocked by a slow transport - Given a connected MCP adapter with a slow discovery transport taking 0.3 seconds + Given a connected MCP adapter with a slow discovery transport taking 0.15 seconds When I call discover_tools concurrently from 3 threads Then all 3 concurrent discoveries should succeed - And the total discovery wall-clock time should be less than 0.9 seconds + And the total discovery wall-clock time should be less than 0.40 seconds Scenario: Lock is not held during invoke transport call Given a connected MCP adapter with a lock-checking transport tool "check_lock" -- 2.52.0 From 236499f0864e2f2331ab00a63a7c9d99d1ccb6f9 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 23 Apr 2026 16:48:03 +0000 Subject: [PATCH 5/8] fix(tests): replace timing-based concurrency assertions with deterministic counter approach --- ...tdd_mcp_adapter_rlock_concurrency_steps.py | 109 +++++++++++++++--- .../tdd_mcp_adapter_rlock_concurrency.feature | 8 +- 2 files changed, 97 insertions(+), 20 deletions(-) diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py index d574b61fb..cd49feb30 100644 --- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -54,6 +54,46 @@ class SlowMCPTransport(MockMCPTransport): return super().call(method, params) +class ConcurrencyTrackingTransport(MockMCPTransport): + """Transport that tracks the maximum number of concurrent in-flight calls. + + Uses a threading.Lock-protected counter to record how many calls are + executing simultaneously. The ``max_concurrent`` attribute reflects the + peak concurrency observed across all calls. + + If the adapter holds its RLock during transport calls, only one call can + be in-flight at a time (max_concurrent == 1). If the lock is released + before the transport call, multiple calls can overlap (max_concurrent > 1). + """ + + def __init__( + self, + delay: float, + tools: list[dict[str, Any]] | None = None, + ) -> None: + super().__init__(tools=tools) + self._delay = delay + self._counter_lock = threading.Lock() + self._current_concurrent: int = 0 + self.max_concurrent: int = 0 + + def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + with self._counter_lock: + self._current_concurrent += 1 + if self._current_concurrent > self.max_concurrent: + self.max_concurrent = self._current_concurrent + + # Use the real time.sleep (not the capped test version) to ensure + # the delay is meaningful for concurrency testing. + _real_sleep = getattr(time, "_original_sleep", time.sleep) + _real_sleep(self._delay) + + with self._counter_lock: + self._current_concurrent -= 1 + + return super().call(method, params) + + class LockCheckingInvokeTransport(MockMCPTransport): """Transport that checks whether the adapter's RLock is held during call().""" @@ -140,6 +180,20 @@ def step_slow_transport_tool(context: Context, tool_name: str, delay: float) -> context.mcp_adapter.discover_tools() +@given( + 'a connected MCP adapter with a concurrency-tracking transport tool "{tool_name}" taking {delay:f} seconds' +) +def step_concurrency_tracking_transport_tool( + context: Context, tool_name: str, delay: float +) -> None: + tools = [_mock_tool(tool_name)] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + context.mcp_transport = ConcurrencyTrackingTransport(delay=delay, tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=context.mcp_transport) + context.mcp_adapter.connect() + context.mcp_adapter.discover_tools() + + @given( "a connected MCP adapter with a slow discovery transport taking {delay:f} seconds" ) @@ -151,6 +205,19 @@ def step_slow_discovery_transport(context: Context, delay: float) -> None: context.mcp_adapter.connect() +@given( + "a connected MCP adapter with a concurrency-tracking discovery transport taking {delay:f} seconds" +) +def step_concurrency_tracking_discovery_transport( + context: Context, delay: float +) -> None: + tools = [_mock_tool("tool_0"), _mock_tool("tool_1")] + config = MCPServerConfig(name="test-server", transport="stdio", command="echo") + context.mcp_transport = ConcurrencyTrackingTransport(delay=delay, tools=tools) + context.mcp_adapter = MCPToolAdapter(config=config, transport=context.mcp_transport) + context.mcp_adapter.connect() + + @given('a connected MCP adapter with a lock-checking transport tool "{tool_name}"') def step_lock_checking_invoke_transport(context: Context, tool_name: str) -> None: tools = [_mock_tool(tool_name)] @@ -195,7 +262,6 @@ def step_lock_checking_discovery_transport(context: Context) -> None: @when('I invoke "{tool_name}" concurrently from {count:d} threads') def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None: results: list[MCPToolResult] = [] - start = time.monotonic() with ThreadPoolExecutor(max_workers=count) as executor: futures = [ @@ -206,14 +272,12 @@ def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None results.append(future.result()) context.concurrent_invoke_results = results - context.concurrent_wall_clock = time.monotonic() - start @when("I call discover_tools concurrently from {count:d} threads") def step_concurrent_discover(context: Context, count: int) -> None: results: list[Any] = [] errors: list[Exception] = [] - start = time.monotonic() with ThreadPoolExecutor(max_workers=count) as executor: futures = [ @@ -227,7 +291,6 @@ def step_concurrent_discover(context: Context, count: int) -> None: context.concurrent_discover_results = results context.concurrent_discover_errors = errors - context.concurrent_discover_wall_clock = time.monotonic() - start # --------------------------------------------------------------- @@ -243,12 +306,19 @@ def step_all_invocations_succeed(context: Context, count: int) -> None: assert result.success, f"Invocation {i} failed: {result.error}" -@then("the total wall-clock time should be less than {max_seconds:f} seconds") -def step_wall_clock_less_than(context: Context, max_seconds: float) -> None: - actual = context.concurrent_wall_clock - assert actual < max_seconds, ( - f"Total wall-clock time {actual:.3f}s >= {max_seconds}s. " - f"Concurrent calls appear to be serialized (lock held during transport call)." +@then( + "at least {min_concurrent:d} invocations should have been in-flight simultaneously" +) +def step_at_least_n_invocations_concurrent( + context: Context, min_concurrent: int +) -> None: + transport = context.mcp_transport + actual = transport.max_concurrent + assert actual >= min_concurrent, ( + f"Expected at least {min_concurrent} concurrent in-flight invocations, " + f"but observed max_concurrent={actual}. " + f"The adapter's RLock appears to be held during transport calls, " + f"serializing concurrent operations." ) @@ -260,12 +330,19 @@ def step_all_discoveries_succeed(context: Context, count: int) -> None: assert len(results) == count, f"Expected {count} results, got {len(results)}" -@then("the total discovery wall-clock time should be less than {max_seconds:f} seconds") -def step_discovery_wall_clock_less_than(context: Context, max_seconds: float) -> None: - actual = context.concurrent_discover_wall_clock - assert actual < max_seconds, ( - f"Total discovery wall-clock time {actual:.3f}s >= {max_seconds}s. " - f"Concurrent calls appear to be serialized (lock held during transport call)." +@then( + "at least {min_concurrent:d} discoveries should have been in-flight simultaneously" +) +def step_at_least_n_discoveries_concurrent( + context: Context, min_concurrent: int +) -> None: + transport = context.mcp_transport + actual = transport.max_concurrent + assert actual >= min_concurrent, ( + f"Expected at least {min_concurrent} concurrent in-flight discoveries, " + f"but observed max_concurrent={actual}. " + f"The adapter's RLock appears to be held during transport calls, " + f"serializing concurrent operations." ) diff --git a/features/tdd_mcp_adapter_rlock_concurrency.feature b/features/tdd_mcp_adapter_rlock_concurrency.feature index 8129b0ddc..b284118a7 100644 --- a/features/tdd_mcp_adapter_rlock_concurrency.feature +++ b/features/tdd_mcp_adapter_rlock_concurrency.feature @@ -13,16 +13,16 @@ Feature: MCPToolAdapter releases RLock during transport calls # ------------------------------------------------------------------- Scenario: Concurrent invoke calls are not blocked by a slow transport - Given a connected MCP adapter with a slow transport tool "slow_op" taking 0.15 seconds + Given a connected MCP adapter with a concurrency-tracking transport tool "slow_op" taking 0.05 seconds When I invoke "slow_op" concurrently from 3 threads Then all 3 concurrent invocations should succeed - And the total wall-clock time should be less than 0.40 seconds + And at least 2 invocations should have been in-flight simultaneously Scenario: Concurrent discover_tools calls are not blocked by a slow transport - Given a connected MCP adapter with a slow discovery transport taking 0.15 seconds + Given a connected MCP adapter with a concurrency-tracking discovery transport taking 0.05 seconds When I call discover_tools concurrently from 3 threads Then all 3 concurrent discoveries should succeed - And the total discovery wall-clock time should be less than 0.40 seconds + And at least 2 discoveries should have been in-flight simultaneously Scenario: Lock is not held during invoke transport call Given a connected MCP adapter with a lock-checking transport tool "check_lock" -- 2.52.0 From 2240a028c8f4cf0bfa570db903e05085d1cd06f7 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 23 Apr 2026 23:13:29 +0000 Subject: [PATCH 6/8] fix(tests): replace ThreadPoolExecutor with threading.Thread in RLock concurrency steps ThreadPoolExecutor holds internal locks (e.g. _global_shutdown_lock) that can be in a locked state after fork(). The behave-parallel runner uses multiprocessing.Pool with the fork start method, so forked worker processes inherit these locked states and deadlock when trying to create or shut down a ThreadPoolExecutor. Replace both concurrent step implementations with direct threading.Thread usage, following the pattern established in context_tier_thread_safety_steps.py and other step files throughout the codebase. This eliminates the fork+lock deadlock while preserving the deterministic concurrency-counter approach for proving the RLock is released during transport calls. --- ...tdd_mcp_adapter_rlock_concurrency_steps.py | 61 +++++++++++++------ 1 file changed, 43 insertions(+), 18 deletions(-) diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py index cd49feb30..b4dca4027 100644 --- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -11,7 +11,6 @@ from __future__ import annotations import threading import time -from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from behave import given, then, when @@ -261,34 +260,60 @@ def step_lock_checking_discovery_transport(context: Context) -> None: @when('I invoke "{tool_name}" concurrently from {count:d} threads') def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None: - results: list[MCPToolResult] = [] + # Use threading.Thread directly instead of ThreadPoolExecutor to avoid + # deadlocks in forked worker processes (behave-parallel uses fork). + # ThreadPoolExecutor holds internal locks that can be in a locked state + # after fork(), causing child processes to deadlock on acquisition. + results: list[MCPToolResult | None] = [None] * count + errors: list[BaseException | None] = [None] * count + errors_lock = threading.Lock() - with ThreadPoolExecutor(max_workers=count) as executor: - futures = [ - executor.submit(context.mcp_adapter.invoke, tool_name, {}) - for _ in range(count) - ] - for future in as_completed(futures): - results.append(future.result()) + def _invoke(idx: int) -> None: + try: + results[idx] = context.mcp_adapter.invoke(tool_name, {}) + except BaseException as exc: + with errors_lock: + errors[idx] = exc - context.concurrent_invoke_results = results + threads = [threading.Thread(target=_invoke, args=(i,)) for i in range(count)] + for t in threads: + t.start() + for t in threads: + t.join() + + # Re-raise the first error encountered, if any + for err in errors: + if err is not None: + raise err + + context.concurrent_invoke_results = [r for r in results if r is not None] @when("I call discover_tools concurrently from {count:d} threads") def step_concurrent_discover(context: Context, count: int) -> None: + # Use threading.Thread directly instead of ThreadPoolExecutor to avoid + # deadlocks in forked worker processes (behave-parallel uses fork). + # ThreadPoolExecutor holds internal locks that can be in a locked state + # after fork(), causing child processes to deadlock on acquisition. results: list[Any] = [] errors: list[Exception] = [] + collect_lock = threading.Lock() - with ThreadPoolExecutor(max_workers=count) as executor: - futures = [ - executor.submit(context.mcp_adapter.discover_tools) for _ in range(count) - ] - for future in as_completed(futures): - try: - results.append(future.result()) - except Exception as exc: + def _discover() -> None: + try: + result = context.mcp_adapter.discover_tools() + with collect_lock: + results.append(result) + except Exception as exc: + with collect_lock: errors.append(exc) + threads = [threading.Thread(target=_discover) for _ in range(count)] + for t in threads: + t.start() + for t in threads: + t.join() + context.concurrent_discover_results = results context.concurrent_discover_errors = errors -- 2.52.0 From beaeac165d1645b888e4f80306ee21e14136e425 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Fri, 24 Apr 2026 08:23:39 +0000 Subject: [PATCH 7/8] fix(db): add merge migration to resolve multiple Alembic heads m4_004_schema_parity_resource_decision_checkpoint was added to master after this branch was created, creating a second Alembic head alongside a5_006_action_invariants_unique_constraint. The CI unit_tests job runs on the merge commit (PR branch + master), so both migration files are present, causing create_template_db.py to fail with MultipleHeads. Add a5_007_merge_m4_004_and_a5_006 no-op merge migration to resolve the two heads into a single head. --- .../a5_007_merge_m4_004_and_a5_006.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py diff --git a/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py b/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py new file mode 100644 index 000000000..17dd49a66 --- /dev/null +++ b/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py @@ -0,0 +1,35 @@ +"""Merge m4_004_schema_parity and a5_006_action_invariants heads. + +``m4_004_schema_parity_resource_decision_checkpoint`` branched off +``m4_003_plan_env_columns`` independently of the main migration chain +(m5_001 -> m8_002 -> m6_006 -> m9_001 -> m9_002 -> a5_006), creating two +Alembic heads. This no-op merge migration resolves them into a single +head so that ``alembic upgrade head`` and the template-DB creation +script work correctly. + +Revision ID: a5_007_merge_m4_004_and_a5_006 +Revises: m4_004_schema_parity_resource_decision_checkpoint, + a5_006_action_invariants_unique_constraint +Create Date: 2026-04-24 00:00:00 +""" + +from collections.abc import Sequence + +# revision identifiers, used by Alembic. +revision: str = "a5_007_merge_m4_004_and_a5_006" +down_revision: str | Sequence[str] | None = ( + "m4_004_schema_parity_resource_decision_checkpoint", + "a5_006_action_invariants_unique_constraint", +) +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """No-op merge migration.""" + pass + + +def downgrade() -> None: + """No-op merge migration.""" + pass -- 2.52.0 From 90b06e630861215f20b4d13523a65e0664b97323 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 3 May 2026 00:04:18 +0000 Subject: [PATCH 8/8] fix(db): remove redundant a5_007 merge migration (m4_004 already merged into m8_002 on master) --- .../a5_007_merge_m4_004_and_a5_006.py | 35 ------------------- 1 file changed, 35 deletions(-) delete mode 100644 src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py diff --git a/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py b/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py deleted file mode 100644 index 17dd49a66..000000000 --- a/src/cleveragents/infrastructure/database/migrations/versions/a5_007_merge_m4_004_and_a5_006.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Merge m4_004_schema_parity and a5_006_action_invariants heads. - -``m4_004_schema_parity_resource_decision_checkpoint`` branched off -``m4_003_plan_env_columns`` independently of the main migration chain -(m5_001 -> m8_002 -> m6_006 -> m9_001 -> m9_002 -> a5_006), creating two -Alembic heads. This no-op merge migration resolves them into a single -head so that ``alembic upgrade head`` and the template-DB creation -script work correctly. - -Revision ID: a5_007_merge_m4_004_and_a5_006 -Revises: m4_004_schema_parity_resource_decision_checkpoint, - a5_006_action_invariants_unique_constraint -Create Date: 2026-04-24 00:00:00 -""" - -from collections.abc import Sequence - -# revision identifiers, used by Alembic. -revision: str = "a5_007_merge_m4_004_and_a5_006" -down_revision: str | Sequence[str] | None = ( - "m4_004_schema_parity_resource_decision_checkpoint", - "a5_006_action_invariants_unique_constraint", -) -branch_labels: str | Sequence[str] | None = None -depends_on: str | Sequence[str] | None = None - - -def upgrade() -> None: - """No-op merge migration.""" - pass - - -def downgrade() -> None: - """No-op merge migration.""" - pass -- 2.52.0