Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 19e96ff7bb |
@@ -77,6 +77,12 @@
|
||||
- **Plan Rollback Command** (#8557): Implemented `agents plan rollback <plan-id> [<checkpoint-id>]` for checkpoint-based plan state restoration in Epic #8493. The command restores a plan's sandbox to the state captured at a given checkpoint, discarding all decisions made after that checkpoint. The checkpoint can be specified as an optional positional second argument or via the `--to-checkpoint` named option. Supports `--yes/-y` flag to skip confirmation prompts and `--format/-f` for output format selection (rich/plain/json/yaml). Included with comprehensive BDD test coverage (>= 97%) and spec-aligned output formatting showing rollback summary, changes reverted, impact analysis, and post-rollback state panels.
|
||||
|
||||
### Fixed
|
||||
- **Race condition in ``McpClient.start()`` allows concurrent double initialisation** (#10438):
|
||||
Added ``_state == McpClientState.STARTING`` check inside the ``threading.RLock`` in
|
||||
``start()`` so that concurrent callers see the in-progress state and return immediately,
|
||||
preventing double initialisation of the MCP server connection, resource leaks, and state
|
||||
corruption. TDD regression test added with BDD scenarios covering concurrent and sequential
|
||||
start paths.
|
||||
- **Global CLI options ``--data-dir``, ``--config-path``, and ``-v`` now work correctly**
|
||||
(#6785): These spec-required flags were absent from ``main_callback()`` in
|
||||
``src/cleveragents/cli/main.py``, causing any invocation with these flags to crash
|
||||
|
||||
@@ -40,3 +40,4 @@ Below are some of the specific details of various contributions.
|
||||
* HAL 9000 has contributed database resource types (PostgreSQL, SQLite) with transaction-based sandbox strategy: implemented ``DatabaseResourceHandler`` providing full CRUD operations (`read`, `write`, `delete`, `list_children`) and connection validation with automatic credential masking for PostgreSQL and SQLite backends. Includes ``TransactionSandbox`` infrastructure wired into ``SandboxFactory``, BDD test coverage in ``features/database_resources.feature``, and Robot Framework integration tests in ``robot/database_resources.robot`` (PR #10591 / issue #8608, Epic #8568).
|
||||
* HAL 9000 has contributed the agents plan rollback command (PR #8674 / issue #8557): implemented checkpoint-based plan state restoration with the `agents plan rollback <plan-id> [<checkpoint-id>]` CLI command as part of Epic #8493, enabling plans to be restored to previous checkpoints, discarding post-checkpoint decisions, and resuming execution from the rolled-back state. Supported by `--yes/-y`, `--to-checkpoint`, and `--format/-f` flags. Includes comprehensive BDD test coverage (>= 97%) for rollback, decision discarding, and plan resume functionality.
|
||||
* HAL 9000 has contributed the PyYAML security upgrade (PR #11012 / issue #9055): added `pyyaml>=6.0.3` dependency constraint to address known YAML parsing vulnerabilities.
|
||||
* HAL 9000 has contributed the AutoDebug node state mutation fix (#10496): fixed _analyze_error, _generate_fix, and _validate_fix in src/cleveragents/agents/graphs/auto_debug.py to return partial update dicts per LangGraph s node contract, preventing duplicate state entries and checkpoint inconsistencies.
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""Counting MCP transport for race-condition testing.
|
||||
|
||||
This module provides :class:`CountingMCPTransport`, a test double that wraps a
|
||||
``MockMCPTransport`` and records how many times ``connect()`` and
|
||||
``discover_tools()`` (tools/list) are called. It uses thread-safe counters so
|
||||
that concurrent scenarios can assert the exact call counts expected when fixing
|
||||
race conditions in ``McpClient.start()`` (issue #10438).
|
||||
|
||||
See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from cleveragents.mcp.adapter import MCPServerConfig, MCPTransport
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
|
||||
class CountingMCPTransport(MCPTransport):
|
||||
"""Transport that counts ``connect()`` and ``discover_tools()`` calls.
|
||||
|
||||
Wraps an inner transport (typically :class:`MockMCPTransport`) and
|
||||
increments thread-safe counters for each ``connect()`` call and every
|
||||
``tools/list`` method call on ``call()``. Used to verify that concurrent
|
||||
:meth:`~McpClient.start` invocations produce exactly one initialisation
|
||||
sequence.
|
||||
"""
|
||||
|
||||
def __init__(self, tools: list[dict[str, Any]]) -> None:
|
||||
from features.mocks.mock_mcp_transport import MockMCPTransport # noqa: PLC0415
|
||||
|
||||
self._inner = MockMCPTransport(tools=tools)
|
||||
self._lock: threading.Lock = threading.Lock()
|
||||
self.connect_count: int = 0
|
||||
self.discover_count: int = 0
|
||||
|
||||
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
self.connect_count += 1
|
||||
return self._inner.connect(config)
|
||||
|
||||
def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||
if method == "tools/list":
|
||||
with self._lock:
|
||||
self.discover_count += 1
|
||||
return self._inner.call(method, params)
|
||||
|
||||
def close(self) -> None:
|
||||
self._inner.close()
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Step definitions for TDD issue #10438 — McpClient.start() race condition.
|
||||
|
||||
This module provides the Behave step implementations for the TDD bug-capture
|
||||
test that proves and verifies the fix for bug #10438.
|
||||
|
||||
The bug was a race condition in ``McpClient.start()``: the method released the
|
||||
``threading.RLock`` after checking ``_started`` but before calling
|
||||
``connect()`` and ``discover_tools()``. Concurrent callers could both pass
|
||||
the idempotency check and initialise the MCP server connection multiple times,
|
||||
causing resource leaks and state corruption.
|
||||
|
||||
The fix adds ``_state == McpClientState.STARTING`` to the guard inside the
|
||||
lock so that a second concurrent caller returns immediately without calling
|
||||
``connect()`` or ``discover_tools()`` again.
|
||||
|
||||
Race detection uses a counting transport that records how many times
|
||||
``connect()`` and ``discover_tools()`` are called. A ``threading.Barrier``
|
||||
synchronises the threads so they all enter ``start()`` at approximately the
|
||||
same instant, maximising the chance of the race manifesting.
|
||||
|
||||
See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from 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, McpClientState
|
||||
from features.mocks.counting_mcp_transport import CountingMCPTransport
|
||||
|
||||
|
||||
|
||||
# ── Given ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given("an MCP client configured for concurrent start testing")
|
||||
def step_mcp_client_for_concurrent_start(context: Context) -> None:
|
||||
"""Create an MCP client with a counting transport for race testing.
|
||||
|
||||
The counting transport records how many times ``connect()`` and
|
||||
``discover_tools()`` are called so that the test can assert exactly
|
||||
one invocation of each after concurrent ``start()`` calls.
|
||||
"""
|
||||
tools = [{"name": "test_tool", "description": "Test tool", "inputSchema": {}}]
|
||||
transport = CountingMCPTransport(tools=tools)
|
||||
|
||||
config = McpClientConfig(
|
||||
server=MCPServerConfig(name="test-server", transport="stdio", command="echo"),
|
||||
lazy_start=False,
|
||||
idle_timeout_seconds=0,
|
||||
health_check_interval_seconds=0,
|
||||
)
|
||||
client = McpClient(config=config, transport=transport)
|
||||
|
||||
context.mcp_race_client = client
|
||||
context.mcp_race_transport = transport
|
||||
context.mcp_race_thread_errors: list[Exception] = []
|
||||
context.mcp_race_thread_count = 0
|
||||
|
||||
|
||||
# ── When ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@when("{n:d} threads call start() simultaneously through a barrier")
|
||||
def step_concurrent_start(context: Context, n: int) -> None:
|
||||
"""Launch *n* threads that all call ``start()`` simultaneously.
|
||||
|
||||
A ``threading.Barrier`` synchronises the threads so they all enter
|
||||
``start()`` at (approximately) the same instant, maximising the chance
|
||||
of the race manifesting.
|
||||
|
||||
The counting transport records how many times ``connect()`` and
|
||||
``discover_tools()`` are called.
|
||||
"""
|
||||
client: McpClient = context.mcp_race_client
|
||||
barrier = threading.Barrier(n)
|
||||
errors: list[Exception] = []
|
||||
collect_lock = threading.Lock()
|
||||
|
||||
def worker() -> None:
|
||||
"""Thread worker: wait on barrier, then call start()."""
|
||||
try:
|
||||
barrier.wait(timeout=10)
|
||||
client.start()
|
||||
except Exception as exc:
|
||||
with collect_lock:
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=worker, daemon=True) for _ in range(n)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(timeout=30)
|
||||
|
||||
# Verify no threads are still alive after join timeout
|
||||
for t in threads:
|
||||
assert not t.is_alive(), (
|
||||
f"Thread {t.name} is still alive after join timeout — "
|
||||
f"possible deadlock in start()"
|
||||
)
|
||||
|
||||
barrier_errors = [
|
||||
err for err in errors if isinstance(err, threading.BrokenBarrierError)
|
||||
]
|
||||
assert not barrier_errors, (
|
||||
"Barrier synchronization failed while coordinating concurrent "
|
||||
f"start() calls. Barrier errors: {barrier_errors}"
|
||||
)
|
||||
|
||||
context.mcp_race_thread_errors = errors
|
||||
context.mcp_race_thread_count = n
|
||||
|
||||
|
||||
@when("I call start() sequentially {n:d} times")
|
||||
def step_sequential_start(context: Context, n: int) -> None:
|
||||
"""Call ``start()`` sequentially *n* times on the same client."""
|
||||
client: McpClient = context.mcp_race_client
|
||||
for _ in range(n):
|
||||
client.start()
|
||||
|
||||
context.mcp_race_thread_errors = []
|
||||
context.mcp_race_thread_count = n
|
||||
|
||||
|
||||
# ── Then ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@then("connect should have been called exactly once")
|
||||
def step_connect_called_once(context: Context) -> None:
|
||||
"""Assert that ``connect()`` was called exactly once.
|
||||
|
||||
With the bug present (no STARTING state guard), multiple threads would
|
||||
pass the ``_started`` check and each call ``connect()``, so the count
|
||||
would be > 1.
|
||||
|
||||
With the fix applied, only the first thread calls ``connect()`` and
|
||||
subsequent concurrent callers see ``_state == STARTING`` and return.
|
||||
"""
|
||||
transport: CountingMCPTransport = context.mcp_race_transport
|
||||
count = transport.connect_count
|
||||
assert count == 1, (
|
||||
f"connect() was called {count} time(s), expected exactly 1. "
|
||||
f"This indicates the race condition in McpClient.start() is present — "
|
||||
f"multiple threads bypassed the idempotency guard. "
|
||||
f"Thread errors (if any): {context.mcp_race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("discover_tools should have been called exactly once")
|
||||
def step_discover_tools_called_once(context: Context) -> None:
|
||||
"""Assert that ``discover_tools()`` was called exactly once.
|
||||
|
||||
With the bug present, multiple threads would call ``discover_tools()``
|
||||
concurrently, so the count would be > 1.
|
||||
|
||||
With the fix applied, only the first thread calls ``discover_tools()``
|
||||
and subsequent concurrent callers return early.
|
||||
"""
|
||||
transport: CountingMCPTransport = context.mcp_race_transport
|
||||
count = transport.discover_count
|
||||
assert count == 1, (
|
||||
f"discover_tools() was called {count} time(s), expected exactly 1. "
|
||||
f"This indicates the race condition in McpClient.start() is present — "
|
||||
f"multiple threads bypassed the idempotency guard. "
|
||||
f"Thread errors (if any): {context.mcp_race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("the MCP client state should be running after concurrent starts")
|
||||
def step_client_state_running_after_concurrent(context: Context) -> None:
|
||||
"""Assert the client is in RUNNING state after concurrent start() calls."""
|
||||
client: McpClient = context.mcp_race_client
|
||||
actual = client.state
|
||||
assert actual == McpClientState.RUNNING, (
|
||||
f"Expected client state to be '{McpClientState.RUNNING}', got '{actual}'. "
|
||||
f"Thread errors (if any): {context.mcp_race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("all concurrent start threads should have completed without error")
|
||||
def step_all_threads_no_error(context: Context) -> None:
|
||||
"""Assert that all concurrent start() threads completed without error."""
|
||||
errors = context.mcp_race_thread_errors
|
||||
assert not errors, f"Expected no thread errors but got {len(errors)}: {errors}"
|
||||
@@ -0,0 +1,39 @@
|
||||
# This test captures bug #10438 — McpClient.start() race condition allowing
|
||||
# concurrent double initialisation.
|
||||
#
|
||||
# The fix adds a check for ``_state == McpClientState.STARTING`` inside the
|
||||
# lock so that a second concurrent caller sees the in-progress state and
|
||||
# returns immediately, preventing double initialisation.
|
||||
#
|
||||
# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
|
||||
|
||||
@tdd_issue @tdd_issue_10438
|
||||
Feature: TDD Issue #10438 — McpClient.start() race condition allows concurrent double initialisation
|
||||
McpClient.start() previously released the threading.RLock after checking
|
||||
``_started`` but before calling ``connect()`` and ``discover_tools()``.
|
||||
Concurrent callers could both pass the idempotency check and initialise
|
||||
the MCP server connection multiple times, causing resource leaks and state
|
||||
corruption.
|
||||
|
||||
The fix adds ``_state == McpClientState.STARTING`` to the guard inside the
|
||||
lock so that a second concurrent caller returns immediately.
|
||||
|
||||
Scenario: Concurrent start() calls result in exactly one connect() invocation
|
||||
Given an MCP client configured for concurrent start testing
|
||||
When 5 threads call start() simultaneously through a barrier
|
||||
Then connect should have been called exactly once
|
||||
And discover_tools should have been called exactly once
|
||||
And the MCP client state should be running after concurrent starts
|
||||
|
||||
Scenario: Concurrent start() calls leave the client in RUNNING state
|
||||
Given an MCP client configured for concurrent start testing
|
||||
When 10 threads call start() simultaneously through a barrier
|
||||
Then the MCP client state should be running after concurrent starts
|
||||
And all concurrent start threads should have completed without error
|
||||
|
||||
Scenario: Sequential start() calls remain idempotent after the fix
|
||||
Given an MCP client configured for concurrent start testing
|
||||
When I call start() sequentially 3 times
|
||||
Then connect should have been called exactly once
|
||||
And discover_tools should have been called exactly once
|
||||
And the MCP client state should be running after concurrent starts
|
||||
@@ -207,7 +207,7 @@ class McpClient:
|
||||
If the MCP server cannot be reached.
|
||||
"""
|
||||
with self._lock:
|
||||
if self._started:
|
||||
if self._started or self._state == McpClientState.STARTING:
|
||||
return
|
||||
self._state = McpClientState.STARTING
|
||||
|
||||
@@ -263,7 +263,7 @@ class McpClient:
|
||||
If the client is not started and lazy_start is disabled.
|
||||
"""
|
||||
with self._lock:
|
||||
if self._started:
|
||||
if self._started or self._state == McpClientState.STARTING:
|
||||
return
|
||||
if not self._config.lazy_start:
|
||||
msg = (
|
||||
|
||||
Reference in New Issue
Block a user