From 936864418ff9014b31e09e550beff15e7165aedb Mon Sep 17 00:00:00 2001 From: CleverThis Date: Tue, 5 May 2026 14:41:17 +0000 Subject: [PATCH] fix(cli): restore regression fixes for mcp adapter, migration runner, and test files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restores several regressions introduced in earlier commits on this branch: 1. Restored check_same_thread=False for SQLite in migration_runner.py get_current_revision() — prevents threading errors when called from non-main threads (regression from master). 2. Restored RLock release-before-transport-call pattern in mcp/adapter.py discover_tools() and invoke() — prevents deadlocks when concurrent operations are blocked by slow network I/O (bug #10512 fix). 3. Restored tdd_mcp_adapter_rlock_concurrency.feature and step definitions that verify the RLock fix — these were incorrectly deleted. 4. Restored consolidated_misc.feature scenario for check_same_thread=False and migration_runner_steps.py step definition — incorrectly deleted. 5. Restored session_cli.feature scenarios for full ULID display and session_cli_steps.py step definitions — incorrectly deleted. 6. Restored session.py to use full session IDs (not truncated [:8]). 7. Restored llm_trace_repository.py UnitOfWork mode with explicit session parameter support — incorrectly simplified. Closes #8623 --- features/consolidated_misc.feature | 6 + features/session_cli.feature | 22 + features/steps/llm_trace_steps.py | 11 +- features/steps/migration_runner_steps.py | 11 + features/steps/session_cli_steps.py | 144 ++++++- ...tdd_mcp_adapter_rlock_concurrency_steps.py | 389 ++++++++++++++++++ .../tdd_mcp_adapter_rlock_concurrency.feature | 52 +++ src/cleveragents/cli/commands/session.py | 6 +- .../database/llm_trace_repository.py | 38 +- .../database/migration_runner.py | 8 +- src/cleveragents/mcp/adapter.py | 147 ++++--- 11 files changed, 758 insertions(+), 76 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/consolidated_misc.feature b/features/consolidated_misc.feature index f1caf6b8e..7649d56ea 100644 --- a/features/consolidated_misc.feature +++ b/features/consolidated_misc.feature @@ -1714,6 +1714,12 @@ Feature: Consolidated Misc And the temporary connection should be closed afterward + Scenario: get_current_revision uses check_same_thread=False for SQLite engines + Given a migration runner configured for "sqlite:///:memory:" + When I request the current revision from the database + Then the SQLite engine for get_current_revision should use check_same_thread=False + + Scenario: File-based SQLite database directory is created if missing Given a migration runner configured for "sqlite:///tmp/test-db/mydb.db" When I initialize or upgrade a file-based SQLite database diff --git a/features/session_cli.feature b/features/session_cli.feature index b363e03ea..215771676 100644 --- a/features/session_cli.feature +++ b/features/session_cli.feature @@ -44,6 +44,28 @@ Feature: Session CLI commands When I run session CLI list with --format json Then the session CLI JSON list entries should match the documented contract + Scenario: List sessions displays full 26-character ULIDs in Rich table + Given there are mocked existing sessions + When I run session CLI list + Then the session CLI rich table should display full session ULIDs + + Scenario: List sessions summary panel shows full ULIDs for unnamed sessions + Given there are mocked existing sessions + When I run session CLI list + Then the session CLI summary panel should contain full session ULIDs + + Scenario: List sessions summary panel shows session names for named sessions + Given there are mocked existing named sessions + When I run session CLI list + Then the session CLI summary panel should show session names + + Scenario: Full session ID from list output works with session tell + Given there are mocked existing sessions + When I run session CLI list + And I capture the first session full ULID from the output + And I run session CLI tell with the full session ID and prompt "Hello from list" + Then the session CLI tell should succeed + # Show command tests Scenario: Show session with valid ID Given there is a mocked session with messages diff --git a/features/steps/llm_trace_steps.py b/features/steps/llm_trace_steps.py index eed9f756d..67e24da83 100644 --- a/features/steps/llm_trace_steps.py +++ b/features/steps/llm_trace_steps.py @@ -809,6 +809,9 @@ class _BrokenSession: def rollback(self) -> None: pass + def close(self) -> None: + pass + def query(self, *_args: Any, **_kwargs: Any) -> Any: raise SQLAlchemyDatabaseError("mock", {}, Exception("broken")) @@ -985,8 +988,10 @@ def step_save_with_spy(context: Context) -> None: object.__setattr__(real_session, "flush", spy_flush) object.__setattr__(real_session, "commit", spy_commit) + # Pass the session explicitly to test the UoW path: save() must flush + # but must NOT commit (the caller owns the transaction boundary). repo = LLMTraceRepository(session_factory=lambda: real_session) - repo.save(context.trace) + repo.save(context.trace, session=real_session) # Commit so the data is visible for subsequent queries object.__setattr__(real_session, "commit", original_commit) real_session.commit() @@ -1030,7 +1035,9 @@ def step_save_in_uow_rollback(context: Context) -> None: session = context.uow_session_factory() repo = LLMTraceRepository(session_factory=lambda: session) try: - repo.save(context.trace) + # Pass the session explicitly to use UoW mode: save() flushes but + # does NOT commit, so the caller's rollback can undo the change. + repo.save(context.trace, session=session) # Simulate a subsequent failure that triggers rollback raise RuntimeError("Simulated failure after save") except RuntimeError: diff --git a/features/steps/migration_runner_steps.py b/features/steps/migration_runner_steps.py index ec0018297..beef1f61a 100644 --- a/features/steps/migration_runner_steps.py +++ b/features/steps/migration_runner_steps.py @@ -316,6 +316,17 @@ def step_then_temp_connection_closed(context) -> None: assert context.current_rev_fake_engine.connections[0].exit_called is True +@then("the SQLite engine for get_current_revision should use check_same_thread=False") +def step_then_get_current_revision_check_same_thread(context) -> None: + _url, kwargs = context.current_rev_create_call + assert "connect_args" in kwargs, ( + "Expected connect_args to be passed to create_engine for SQLite" + ) + assert kwargs["connect_args"].get("check_same_thread") is False, ( + "Expected check_same_thread=False in connect_args for SQLite engine" + ) + + @when("I initialize or upgrade a file-based SQLite database") def step_when_init_file_based_sqlite(context) -> None: import shutil diff --git a/features/steps/session_cli_steps.py b/features/steps/session_cli_steps.py index 779643ffb..4c33d7e9f 100644 --- a/features/steps/session_cli_steps.py +++ b/features/steps/session_cli_steps.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import re import tempfile from datetime import datetime from typing import Any @@ -160,14 +161,34 @@ def step_existing_sessions(context: Context) -> None: context.mock_service.list.return_value = sessions +@given("there are mocked existing named sessions") +def step_existing_named_sessions(context: Context) -> None: + """Set up mocked sessions with names for Summary panel name-display test.""" + sessions = [ + _make_session( + session_id=_SESSION_ID, + actor_name="openai/gpt-4", + messages=[_make_message(sequence=0)], + ), + _make_session(session_id=_SESSION_ID_2), + ] + sessions[0].name = "weekly-planning" + sessions[1].name = "refactor-sprint" + context.mock_service.list.return_value = sessions + + @when("I run session CLI list") def step_list(context: Context) -> None: - context.result = context.runner.invoke(session_app, ["list"]) + context.result = context.runner.invoke( + session_app, ["list"], env={"COLUMNS": "200"} + ) @when("I run session CLI list with --format json") def step_list_json(context: Context) -> None: - context.result = context.runner.invoke(session_app, ["list", "--format", "json"]) + context.result = context.runner.invoke( + session_app, ["list", "--format", "json"], env={"COLUMNS": "200"} + ) @then("the session CLI should show all sessions in a table") @@ -176,6 +197,125 @@ def step_list_shows_table(context: Context) -> None: assert "Sessions" in context.result.output +@then("the session CLI rich table should display full session ULIDs") +def step_list_rich_table_full_ulids(context: Context) -> None: + """Verify the Rich table displays the full 26-character session ULIDs.""" + assert context.result.exit_code == 0 + output = context.result.output + # Restrict assertion to the table region (before the Summary panel) so + # that a regression back to 8-char truncation is not masked by the full + # ULID appearing elsewhere (e.g. in the Summary panel). + summary_idx = output.find("Summary") + table_output = output[:summary_idx] if summary_idx != -1 else output + assert _SESSION_ID in table_output, ( + f"Full ULID {_SESSION_ID} not found in Rich table output:\n" + f"{context.result.output}" + ) + assert _SESSION_ID_2 in table_output, ( + f"Full ULID {_SESSION_ID_2} not found in Rich table output:\n" + f"{context.result.output}" + ) + # Negative guard: ensure the table contains full 26-character ULIDs, + # not the old 8-character truncated form. A standalone 8-char prefix + # (not as part of the full ULID) would indicate the [:8] slice was not + # removed. + table_without_full_ids = table_output.replace(_SESSION_ID, "").replace( + _SESSION_ID_2, "" + ) + assert _SESSION_ID[:8] not in table_without_full_ids, ( + f"Truncated 8-char ID {_SESSION_ID[:8]} found in Rich table output:\n" + f"{context.result.output}" + ) + assert _SESSION_ID_2[:8] not in table_without_full_ids, ( + f"Truncated 8-char ID {_SESSION_ID_2[:8]} found in Rich table output:\n" + f"{context.result.output}" + ) + + +@then("the session CLI summary panel should contain full session ULIDs") +def step_list_summary_full_ulids(context: Context) -> None: + """Verify the Summary panel shows full ULIDs for unnamed sessions.""" + assert context.result.exit_code == 0 + output = context.result.output + assert "Summary" in output, f"Summary panel not found in output:\n{output}" + # Extract the Summary panel region to avoid a false pass from the Rich + # table also containing the same full ULIDs. + summary_idx = output.find("Summary") + summary_output = output[summary_idx:] + # Both Most Recent and Oldest entries should display full ULIDs when + # sessions are unnamed. Asserting only one ID would allow a regression + # that re-introduced [:8] on one fallback path while keeping the other + # intact to pass the test undetected. + assert _SESSION_ID in summary_output, ( + f"Full ULID {_SESSION_ID} not found in Summary panel:\n{output}" + ) + assert _SESSION_ID_2 in summary_output, ( + f"Full ULID {_SESSION_ID_2} not found in Summary panel:\n{output}" + ) + + +@then("the session CLI summary panel should show session names") +def step_list_summary_shows_names(context: Context) -> None: + """Verify the Summary panel shows session names (not ULIDs) for named sessions.""" + assert context.result.exit_code == 0 + output = context.result.output + summary_idx = output.find("Summary") + assert summary_idx != -1, f"Summary panel not found in output:\n{output}" + summary_output = output[summary_idx:] + assert "weekly-planning" in summary_output, ( + f"'weekly-planning' not found in Summary panel:\n{output}" + ) + assert "refactor-sprint" in summary_output, ( + f"'refactor-sprint' not found in Summary panel:\n{output}" + ) + # The Summary should NOT show the raw ULID when session names are present + assert _SESSION_ID not in summary_output, ( + f"Full ULID {_SESSION_ID} unexpectedly found in Summary panel:\n{output}" + ) + assert _SESSION_ID_2 not in summary_output, ( + f"Full ULID {_SESSION_ID_2} unexpectedly found in Summary panel:\n{output}" + ) + + +@when("I capture the first session full ULID from the output") +def step_capture_first_ulid(context: Context) -> None: + """Parse the first session's full ULID from the output and store it + for subsequent steps (round-trip tell test).""" + assert context.result.exit_code == 0 + output = context.result.output + # Restrict the search to the table region (before the Summary panel) + # so that a ULID appearing in the Summary panel is not accidentally + # captured as the "first" session ID. + summary_idx = output.find("Summary") + search_region = output[:summary_idx] if summary_idx != -1 else output + match = re.search(r"[0-9A-HJKMNP-TV-Z]{26}", search_region) + assert match is not None, ( + f"No 26-character ULID found in table output:\n{search_region}" + ) + context.full_session_id = match.group() + assert len(context.full_session_id) == 26, ( + f"Parsed ULID '{context.full_session_id}' is not 26 characters" + ) + # Sanity check: the captured ID should match one of the known fixture IDs + assert context.full_session_id in (_SESSION_ID, _SESSION_ID_2), ( + f"Captured ULID '{context.full_session_id}' does not match any fixture ID" + ) + + +@when('I run session CLI tell with the full session ID and prompt "{prompt}"') +def step_tell_with_stored_ulid(context: Context, prompt: str) -> None: + """Run session tell using the full ULID stored from session list.""" + session_id = context.full_session_id + context.mock_service.append_message.side_effect = [ + _make_message(MessageRole.USER, prompt, 0), + _make_message(MessageRole.ASSISTANT, f"Acknowledged: {prompt}", 1), + ] + context.result = context.runner.invoke( + session_app, + ["tell", "--session", session_id, prompt], + ) + + # --------------------------------------------------------------------------- # Show # --------------------------------------------------------------------------- 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..b4dca4027 --- /dev/null +++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py @@ -0,0 +1,389 @@ +"""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 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. + _real_sleep = getattr(time, "_original_sleep", time.sleep) + _real_sleep(self._delay) + 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().""" + + 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 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" +) +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 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)] + 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: + # 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() + + def _invoke(idx: int) -> None: + try: + results[idx] = context.mcp_adapter.invoke(tool_name, {}) + except BaseException as exc: + with errors_lock: + errors[idx] = exc + + 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() + + 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 + + +# --------------------------------------------------------------- +# 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( + "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." + ) + + +@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( + "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." + ) + + +@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..b284118a7 --- /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 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 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 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 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" + 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/cli/commands/session.py b/src/cleveragents/cli/commands/session.py index 60e4e591d..30202e6da 100644 --- a/src/cleveragents/cli/commands/session.py +++ b/src/cleveragents/cli/commands/session.py @@ -151,8 +151,8 @@ def _session_list_dict(sessions: list[Session]) -> dict[str, Any]: # Find most recent and oldest sessions if sessions: sorted_sessions = sorted(sessions, key=lambda x: x.updated_at, reverse=True) - most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id[:8] - oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id[:8] + most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id + oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id else: most_recent = None oldest = None @@ -347,7 +347,7 @@ def list_sessions( for s in sessions: table.add_row( - s.session_id[:8], # Truncate ID for readability + s.session_id, # Full ULID for copy-paste compatibility with session tell s.name or "(unnamed)", s.actor_name or "(none)", str(s.message_count), diff --git a/src/cleveragents/infrastructure/database/llm_trace_repository.py b/src/cleveragents/infrastructure/database/llm_trace_repository.py index 70fcfa1ae..a2adc0671 100644 --- a/src/cleveragents/infrastructure/database/llm_trace_repository.py +++ b/src/cleveragents/infrastructure/database/llm_trace_repository.py @@ -31,6 +31,17 @@ class LLMTraceRepository: Uses the session-factory pattern: each public method obtains a session from the factory. Callers are responsible for commit. + + When ``save()`` is called with an explicit ``session`` argument the + repository operates in *UnitOfWork mode*: it flushes the change into + the caller's transaction but does **not** commit or close the session. + The caller (or the enclosing ``UnitOfWork``) is responsible for the + final commit. + + When ``save()`` is called without an explicit ``session`` argument the + repository operates in *standalone mode*: it creates its own session + from the factory, flushes, commits, and closes the session so that the + trace is durably persisted even outside a ``UnitOfWork``. """ def __init__( @@ -46,16 +57,26 @@ class LLMTraceRepository: return self._sf() @database_retry - def save(self, trace: LLMTrace) -> None: + def save(self, trace: LLMTrace, session: Session | None = None) -> None: """Persist a single ``LLMTrace`` row. Args: - trace: The trace to persist. + trace: The trace to persist. Must not be ``None``. + session: Optional external SQLAlchemy session. When provided + the repository flushes into the caller's transaction and + does **not** commit or close the session (UnitOfWork mode). + When omitted the repository creates its own session, commits, + and closes it (standalone mode). Raises: + ValueError: If ``trace`` is ``None``. DatabaseError: On unrecoverable persistence failure. """ - session = self._session() + if trace is None: + raise ValueError("trace must not be None") + + own_session = session is None + s: Session = self._session() if own_session else session try: model = LLMTraceModel( trace_id=trace.trace_id, @@ -77,11 +98,16 @@ class LLMTraceRepository: error=trace.error, timestamp=trace.timestamp.isoformat(), ) - session.add(model) - session.flush() + s.add(model) + s.flush() + if own_session: + s.commit() except (SQLAlchemyDatabaseError, OperationalError) as exc: - session.rollback() + s.rollback() raise DatabaseError(f"Failed to save LLM trace: {exc}") from exc + finally: + if own_session: + s.close() @database_retry def get(self, trace_id: str) -> LLMTrace | None: diff --git a/src/cleveragents/infrastructure/database/migration_runner.py b/src/cleveragents/infrastructure/database/migration_runner.py index c731c1c50..8f8393cb3 100644 --- a/src/cleveragents/infrastructure/database/migration_runner.py +++ b/src/cleveragents/infrastructure/database/migration_runner.py @@ -154,7 +154,13 @@ class MigrationRunner: Returns: Current revision ID or None if no migrations have been applied """ - engine = create_engine(self.database_url) + if self.database_url.startswith("sqlite"): + engine = create_engine( + self.database_url, + connect_args={"check_same_thread": False}, + ) + else: + engine = create_engine(self.database_url) with engine.connect() as connection: context = MigrationContext.configure(connection) return context.get_current_revision() 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,