"""Step definitions for features/tdd_mcp_client_start_race.feature.

TDD issue-capture scenario for #10438: McpClient.start() race condition
allows concurrent double initialization.

McpClient.start() releases the threading.RLock after setting _state to
STARTING but before calling connect() and discover_tools().  Two concurrent
callers can both pass the _started idempotency check (still False) and both
proceed to call connect() and discover_tools() simultaneously.

The test uses @tdd_expected_fail so that CI passes while the bug is unfixed.
The underlying assertion ("connect() called exactly once") fails because the
race allows multiple threads to trigger initialization.  The tag inversion
causes the scenario to be reported as passed.

Race detection uses a threading.Barrier to synchronise threads so they all
enter start() at approximately the same instant, maximising the chance of
the race manifesting.  A counting mock transport records how many times
connect() and discover_tools() are called.

When bug #10438 is fixed (by checking _state == STARTING inside the lock),
the assertions will pass and the @tdd_expected_fail tag must be removed.

See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
"""

from __future__ import annotations

import threading
import time
from typing import Any

from behave import given, then, when
from behave.runner import Context

from cleveragents.mcp.adapter import MCPServerConfig
from cleveragents.mcp.client import McpClient, McpClientConfig
from features.mocks.mock_mcp_transport import MockMCPTransport


class _CountingMockTransport(MockMCPTransport):
    """Mock transport that counts connect() and discover_tools() calls.

    Uses a threading.Lock to make the counters thread-safe so that
    concurrent calls are accurately recorded.
    """

    def __init__(self) -> None:
        tools = [
            {
                "name": "test_tool",
                "description": "Mock tool for race condition test",
                "inputSchema": {},
            }
        ]
        super().__init__(tools=tools)
        self._counter_lock = threading.Lock()
        self.connect_call_count: int = 0
        self.discover_call_count: int = 0
        # Add a small sleep inside connect() to widen the race window,
        # making it more likely that multiple threads enter concurrently.
        self._connect_delay: float = 0.02

    def connect(self, config: MCPServerConfig) -> dict[str, Any]:
        with self._counter_lock:
            self.connect_call_count += 1
        # Sleep outside the counter lock to widen the race window.
        # This gives other threads time to also pass the _started check
        # and enter connect() before the first thread finishes.
        time.sleep(self._connect_delay)
        return super().connect(config)

    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        if method == "tools/list":
            with self._counter_lock:
                self.discover_call_count += 1
        return super().call(method, params)


# ── Given ─────────────────────────────────────────────────────────


@given("an McpClient with a counting mock transport")
def step_mcp_client_counting_transport(context: Context) -> None:
    """Create an McpClient backed by a counting mock transport.

    The client is configured with lazy_start=False, idle_timeout=0, and
    health_check_interval=0 so that only explicit start() calls trigger
    connect() and discover_tools().
    """
    context.counting_transport = _CountingMockTransport()
    server_config = MCPServerConfig(
        name="test-server",
        transport="stdio",
        command="echo",
    )
    client_config = McpClientConfig(
        server=server_config,
        lazy_start=False,
        idle_timeout_seconds=0,
        health_check_interval_seconds=0,
    )
    context.race_client = McpClient(
        config=client_config,
        transport=context.counting_transport,
    )
    context.race_thread_errors: list[Exception] = []
    context.race_threads_completed: int = 0


# ── When ──────────────────────────────────────────────────────────


@when("{n:d} threads call start() concurrently through a barrier")
def step_concurrent_start(context: Context, n: int) -> None:
    """Launch *n* threads that all call start() simultaneously.

    A threading.Barrier synchronises the threads so they all enter
    start() at approximately the same instant, maximising the chance
    of the race condition manifesting.

    With the bug present, multiple threads will pass the _started check
    (still False) and each call connect() and discover_tools().
    With the fix applied, only one thread proceeds past the STARTING
    state check and the others return early.
    """
    barrier = threading.Barrier(n)
    errors: list[Exception] = []
    completed: list[int] = [0]
    collect_lock = threading.Lock()

    def worker() -> None:
        """Thread worker: wait on barrier, then call start()."""
        try:
            barrier.wait(timeout=10)
            context.race_client.start()
            with collect_lock:
                completed[0] += 1
        except Exception as exc:
            with collect_lock:
                errors.append(exc)
                completed[0] += 1

    threads = [threading.Thread(target=worker, daemon=True) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)

    # Verify no threads are still alive after join timeout
    for t in threads:
        assert not t.is_alive(), (
            f"Thread {t.name} is still alive after join timeout — "
            f"possible deadlock in McpClient.start()"
        )

    barrier_errors = [
        err for err in errors if isinstance(err, threading.BrokenBarrierError)
    ]
    assert not barrier_errors, (
        "Barrier synchronization failed while coordinating concurrent "
        "start() calls. This invalidates race setup. "
        f"Barrier errors: {barrier_errors}; all errors: {errors}"
    )

    context.race_thread_errors = errors
    context.race_threads_completed = completed[0]
    context.race_thread_count = n


# ── Then ──────────────────────────────────────────────────────────


@then("connect() should have been called exactly once")
def step_connect_called_once(context: Context) -> None:
    """Assert connect() was called exactly once.

    With the bug present (no STARTING state check inside the lock),
    multiple threads will enter the start() body and each call connect(),
    so the counter will be > 1.

    This assertion therefore FAILS while the bug exists — which is the
    expected behavior for a @tdd_expected_fail test.

    When the fix adds the _state == STARTING check inside the lock, only
    the first thread calls connect() and the count is exactly 1.
    """
    count = context.counting_transport.connect_call_count
    assert count == 1, (
        f"connect() was called {count} time(s), expected exactly 1. "
        f"This confirms the race condition in McpClient.start() — multiple "
        f"threads passed the _started check because the STARTING state is "
        f"not checked inside the lock before proceeding. "
        f"Thread errors (if any): {context.race_thread_errors}"
    )


@then("discover_tools() should have been called exactly once")
def step_discover_tools_called_once(context: Context) -> None:
    """Assert discover_tools() was called exactly once.

    With the bug present, multiple threads call discover_tools() concurrently.
    With the fix, only one thread proceeds to call discover_tools().
    """
    count = context.counting_transport.discover_call_count
    assert count == 1, (
        f"discover_tools() was called {count} time(s), expected exactly 1. "
        f"This confirms the race condition in McpClient.start() — multiple "
        f"threads proceeded past the idempotency check. "
        f"Thread errors (if any): {context.race_thread_errors}"
    )


@then('the client state should be "{expected_state}"')
def step_client_state_is(context: Context, expected_state: str) -> None:
    """Assert the client is in the expected state after concurrent starts."""
    actual = context.race_client.state
    assert actual == expected_state, (
        f"Expected client state '{expected_state}', got '{actual}'. "
        f"Thread errors (if any): {context.race_thread_errors}"
    )


@then("all {n:d} threads should have completed without hanging")
def step_all_threads_completed_no_hang(context: Context, n: int) -> None:
    """Verify that all *n* threads completed (either success or error).

    The total of completed threads must equal the number launched.
    This confirms no threads silently hung or deadlocked.
    """
    total = context.race_threads_completed
    assert total == n, (
        f"Expected {n} threads to complete but only {total} did. "
        "A mismatch indicates a hung thread or possible deadlock in "
        "McpClient.start(). "
        f"Thread errors (if any): {context.race_thread_errors}"
    )
