# -*- coding: utf-8 -*-
"""Location: ./tests/loadtest/locustfile_spin_detector.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Locust load test for detecting CPU spin loop bug (Issue #2360).
This test uses a spike/drop pattern to stress-test session cleanup:
1. Ramp up to high user count (creates many connections/tasks)
2. Drop to 0 users (triggers cleanup of all sessions)
3. Pause to observe CPU behavior (should return to idle)
4. Repeat multiple cycles
The CPU spin loop bug causes workers to consume 100% CPU each when idle
after clients disconnect, due to orphaned asyncio tasks in anyio's
_deliver_cancellation loop.
This is a FULL-FEATURED load test (not simplified) that uses:
- JWT authentication (auto-generated or from MCPGATEWAY_BEARER_TOKEN)
- All user classes from the main locustfile (API, RPC, Admin, FastTime, etc.)
- Entity ID fetching on test start
- Same patterns and weights as load-test-ui
See: https://github.com/IBM/mcp-context-forge/issues/2360
Usage:
    make load-test-spin-detector
    # Or directly:
    cd tests/loadtest && locust -f locustfile_spin_detector.py \
        --host=http://localhost:4444 --headless
"""

# Standard
import logging
import os
import random
import subprocess
import sys
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

# Third-Party
from locust import LoadTestShape, between, constant_throughput, events, tag, task
from locust.contrib.fasthttp import FastHttpUser
from locust.runners import MasterRunner, WorkerRunner

# Configure logging - suppress verbose Locust runner logs by default
# Set LOCUST_VERBOSE=1 to see all logs
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)
logger.setLevel(logging.WARNING)  # Suppress INFO messages from this module

# Flag to control verbose logging - set LOCUST_VERBOSE=1 to see all logs
VERBOSE_LOGGING = os.environ.get("LOCUST_VERBOSE", "0") == "1"

# Worker count - passed from Makefile or auto-detected
# -1 means auto-detect (use all CPUs)
WORKER_COUNT = int(os.environ.get("LOCUST_WORKERS", "-1"))
if WORKER_COUNT == -1:
    WORKER_COUNT = os.cpu_count() or 1


# =============================================================================
# Configuration - Load from .env file and environment variables
# =============================================================================


def _load_env_file() -> dict[str, str]:
    """Load environment variables from .env file.

    Searches for .env file in current directory and parent directories.
    Returns a dict of key-value pairs from the .env file.
    """
    env_vars: dict[str, str] = {}

    # Search for .env file
    search_paths = [
        Path.cwd() / ".env",
        Path.cwd().parent / ".env",
        Path.cwd().parent.parent / ".env",
        Path(__file__).parent.parent.parent / ".env",  # Project root
    ]

    env_file = None
    for path in search_paths:
        if path.exists():
            env_file = path
            break

    if env_file is None:
        return env_vars

    try:
        with open(env_file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                # Skip empty lines and comments
                if not line or line.startswith("#"):
                    continue
                # Handle key=value pairs
                if "=" in line:
                    key, _, value = line.partition("=")
                    key = key.strip()
                    value = value.strip()
                    # Remove quotes if present
                    if value and value[0] in ('"', "'") and value[-1] == value[0]:
                        value = value[1:-1]
                    env_vars[key] = value
    except Exception as e:
        logger.warning(f"Error reading .env file: {e}")

    return env_vars


def _get_config(key: str, default: str = "") -> str:
    """Get configuration value from environment or .env file.

    Priority: Environment variable > .env file > default
    """
    # First check environment variable
    env_value = os.environ.get(key)
    if env_value is not None:
        return env_value

    # Then check .env file
    if key in _ENV_FILE_VARS:
        return _ENV_FILE_VARS[key]

    return default


# Load .env file once at module import
_ENV_FILE_VARS = _load_env_file()

# Authentication settings (from env or .env file)
BEARER_TOKEN = _get_config("MCPGATEWAY_BEARER_TOKEN", "")
BASIC_AUTH_USER = _get_config("BASIC_AUTH_USER", "admin")
BASIC_AUTH_PASSWORD = _get_config("BASIC_AUTH_PASSWORD", "changeme")

# JWT settings for auto-generation (if MCPGATEWAY_BEARER_TOKEN not set)
JWT_SECRET_KEY = _get_config("JWT_SECRET_KEY", "my-test-key-but-now-longer-than-32-bytes")
JWT_ALGORITHM = _get_config("JWT_ALGORITHM", "HS256")
JWT_AUDIENCE = _get_config("JWT_AUDIENCE", "mcpgateway-api")
JWT_ISSUER = _get_config("JWT_ISSUER", "mcpgateway")
# Default to platform admin email for guaranteed authentication
JWT_USERNAME = _get_config("JWT_USERNAME", _get_config("PLATFORM_ADMIN_EMAIL", "admin@example.com"))
# Token expiry in hours - default 8760 (1 year) to avoid expiration during long load tests
JWT_TOKEN_EXPIRY_HOURS = int(_get_config("LOADTEST_JWT_EXPIRY_HOURS", "8760"))

# Test data pools (populated during test setup)
TOOL_IDS: list[str] = []
SERVER_IDS: list[str] = []
GATEWAY_IDS: list[str] = []
RESOURCE_IDS: list[str] = []
PROMPT_IDS: list[str] = []

# Names/URIs for RPC calls
TOOL_NAMES: list[str] = []
RESOURCE_URIS: list[str] = []
PROMPT_NAMES: list[str] = []

# Tools that require arguments - excluded from generic rpc_call_tool
TOOLS_WITH_REQUIRED_ARGS: set[str] = {
    "fast-time-convert-time",
    "fast-time-get-system-time",
    "fast-test-echo",
    "fast-test-get-system-time",
}


# =============================================================================
# ANSI Color Codes
# =============================================================================
class Colors:
    """ANSI color codes for terminal output."""

    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"

    # Foreground colors
    RED = "\033[91m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    BLUE = "\033[94m"
    MAGENTA = "\033[95m"
    CYAN = "\033[96m"
    WHITE = "\033[97m"

    # Background colors
    BG_RED = "\033[41m"
    BG_GREEN = "\033[42m"
    BG_YELLOW = "\033[43m"
    BG_BLUE = "\033[44m"

    @classmethod
    def disable(cls) -> None:
        """Disable colors (for non-TTY output)."""
        for attr in dir(cls):
            if not attr.startswith("_") and isinstance(getattr(cls, attr), str):
                setattr(cls, attr, "")


# Disable colors if not a TTY
if not sys.stdout.isatty():
    Colors.disable()


# =============================================================================
# Logging Setup
# =============================================================================
# Fixed log file path for easy monitoring: tail -f /tmp/spin_detector.log
LOG_FILE = "/tmp/spin_detector.log"
_log_file_handle = None


def _init_log_file() -> None:
    """Initialize the log file."""
    global _log_file_handle
    try:
        _log_file_handle = open(LOG_FILE, "w", encoding="utf-8")
        _log_file_handle.write("# CPU Spin Loop Detector Log\n")
        _log_file_handle.write(f"# Started: {datetime.now().isoformat()}\n")
        _log_file_handle.write("# Issue: https://github.com/IBM/mcp-context-forge/issues/2360\n")
        _log_file_handle.write("#" + "=" * 79 + "\n\n")
        _log_file_handle.flush()
    except Exception as e:
        print(f"{Colors.YELLOW}Warning: Could not create log file: {e}{Colors.RESET}")


def _close_log_file() -> None:
    """Close the log file."""
    global _log_file_handle
    if _log_file_handle:
        _log_file_handle.write(f"\n# Finished: {datetime.now().isoformat()}\n")
        _log_file_handle.close()
        _log_file_handle = None


def log(message: str, to_console: bool = True) -> None:
    """Log a message to both console and file.

    Args:
        message: Message to log (may contain ANSI codes for console).
        to_console: Whether to print to console.
    """
    timestamp = datetime.now().strftime("%H:%M:%S")

    if to_console:
        print(message)
        sys.stdout.flush()

    if _log_file_handle:
        # Strip ANSI codes for file
        clean_msg = message
        for attr in dir(Colors):
            if not attr.startswith("_") and isinstance(getattr(Colors, attr), str):
                clean_msg = clean_msg.replace(getattr(Colors, attr), "")
        _log_file_handle.write(f"[{timestamp}] {clean_msg}\n")
        _log_file_handle.flush()


# =============================================================================
# Docker Stats
# =============================================================================
def get_docker_stats() -> tuple[str, list[tuple[str, float]]]:
    """Get docker stats for gateway containers.

    Returns:
        Tuple of (formatted output string, list of (container_name, cpu_percent)).
    """
    try:
        result = subprocess.run(
            ["docker", "stats", "--no-stream", "--format", "{{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode == 0:
            lines = result.stdout.strip().split("\n")
            gateway_lines = []
            cpu_values = []

            for line in lines:
                if "gateway" in line.lower():
                    parts = line.split("\t")
                    if len(parts) >= 2:
                        name = parts[0]
                        cpu_str = parts[1].replace("%", "")
                        try:
                            cpu = float(cpu_str)
                            cpu_values.append((name, cpu))
                        except ValueError:
                            cpu = 0.0
                        gateway_lines.append(line)

            if gateway_lines:
                header = f"{'CONTAINER':<40} {'CPU %':>10} {'MEMORY':>20}"
                formatted = header + "\n" + "-" * 72 + "\n"
                for line in gateway_lines:
                    parts = line.split("\t")
                    if len(parts) >= 3:
                        formatted += f"{parts[0]:<40} {parts[1]:>10} {parts[2]:>20}\n"
                return formatted, cpu_values
            return "(no gateway containers found)", []
        return f"(docker stats failed: {result.stderr})", []
    except subprocess.TimeoutExpired:
        return "(docker stats timed out)", []
    except FileNotFoundError:
        return "(docker not found)", []
    except Exception as e:
        return f"(error: {e})", []


def format_cpu_status(cpu_values: list[tuple[str, float]], is_pause_phase: bool = False) -> str:
    """Format CPU status with color-coded health indicator.

    Args:
        cpu_values: List of (container_name, cpu_percent) tuples.
        is_pause_phase: If True, high CPU is flagged as FAIL (spin loop detection).
                        If False, high CPU is expected (under load).

    Returns:
        Formatted status string with colors.
    """
    if not cpu_values:
        return f"{Colors.YELLOW}[?] No CPU data{Colors.RESET}"

    max_cpu = max(cpu for _, cpu in cpu_values)
    total_cpu = sum(cpu for _, cpu in cpu_values)

    if is_pause_phase:
        # During pause: CPU should be idle - this is where we detect spin loops
        if max_cpu < 10:
            icon = f"{Colors.GREEN}{Colors.BOLD}[PASS]{Colors.RESET}"
            status = f"{Colors.GREEN}CPU idle - cleanup working correctly{Colors.RESET}"
        elif max_cpu < 50:
            icon = f"{Colors.YELLOW}{Colors.BOLD}[WARN]{Colors.RESET}"
            status = f"{Colors.YELLOW}CPU elevated - may still be cleaning up{Colors.RESET}"
        else:
            icon = f"{Colors.RED}{Colors.BOLD}[FAIL]{Colors.RESET}"
            status = f"{Colors.RED}CPU HIGH during pause - possible spin loop!{Colors.RESET}"
    else:
        # During load: high CPU is expected and normal
        if max_cpu < 10:
            icon = f"{Colors.DIM}[IDLE]{Colors.RESET}"
            status = f"{Colors.DIM}CPU idle (ramping up){Colors.RESET}"
        elif max_cpu < 200:
            icon = f"{Colors.CYAN}[LOAD]{Colors.RESET}"
            status = f"{Colors.CYAN}CPU under load (normal){Colors.RESET}"
        else:
            icon = f"{Colors.GREEN}{Colors.BOLD}[LOAD]{Colors.RESET}"
            status = f"{Colors.GREEN}CPU under heavy load (normal){Colors.RESET}"

    return f"{icon} Total: {total_cpu:.1f}% | Max: {max_cpu:.1f}% - {status}"


# =============================================================================
# Pretty Printing
# =============================================================================
def print_box(title: str, content: str, color: str = Colors.CYAN, width: int = 80) -> None:
    """Print a colored box with title and content.

    Args:
        title: Box title.
        content: Box content.
        color: Color for the border.
        width: Box width.
    """
    top = f"{color}{'=' * width}{Colors.RESET}"
    log(top)
    log(f"{color}{Colors.BOLD}{title.center(width)}{Colors.RESET}")
    log(f"{color}{'=' * width}{Colors.RESET}")
    if content:
        for line in content.split("\n"):
            log(line)


def print_section(title: str, color: str = Colors.BLUE) -> None:
    """Print a section header.

    Args:
        title: Section title.
        color: Color for the header.
    """
    log(f"\n{color}{Colors.BOLD}{title}{Colors.RESET}")
    log(f"{color}{'-' * len(title)}{Colors.RESET}")


# =============================================================================
# Authentication Helpers
# =============================================================================


def _generate_jwt_token() -> str:
    """Generate a JWT token for API authentication.

    Uses PyJWT to create a token with the configured secret and algorithm.
    """
    try:
        from datetime import timedelta, timezone  # pylint: disable=import-outside-toplevel

        import jwt  # pylint: disable=import-outside-toplevel

        jti = str(uuid.uuid4())
        payload = {
            "sub": JWT_USERNAME,
            "exp": datetime.now(timezone.utc) + timedelta(hours=JWT_TOKEN_EXPIRY_HOURS),
            "iat": datetime.now(timezone.utc),
            "aud": JWT_AUDIENCE,
            "iss": JWT_ISSUER,
            "jti": jti,
        }
        token = jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
        return token
    except ImportError:
        logger.warning("PyJWT not installed, falling back to basic auth")
        return ""
    except Exception as e:
        logger.warning(f"Failed to generate JWT token: {e}")
        return ""


# Cache the generated token
_CACHED_TOKEN: str | None = None


def _get_auth_headers() -> dict[str, str]:
    """Get authentication headers.

    Priority:
    1. MCPGATEWAY_BEARER_TOKEN env var (if set)
    2. Auto-generated JWT token (if PyJWT available)
    3. Basic auth fallback
    """
    global _CACHED_TOKEN  # pylint: disable=global-statement
    headers = {"Accept": "application/json"}

    if BEARER_TOKEN:
        headers["Authorization"] = f"Bearer {BEARER_TOKEN}"
    else:
        if _CACHED_TOKEN is None:
            _CACHED_TOKEN = _generate_jwt_token()

        if _CACHED_TOKEN:
            headers["Authorization"] = f"Bearer {_CACHED_TOKEN}"
        else:
            import base64  # pylint: disable=import-outside-toplevel

            credentials = base64.b64encode(f"{BASIC_AUTH_USER}:{BASIC_AUTH_PASSWORD}".encode()).decode()
            headers["Authorization"] = f"Basic {credentials}"
            logger.warning("Using basic auth - REST API endpoints may fail")

    return headers


def _log_auth_mode() -> None:
    """Log which authentication mode the load test will use."""
    headers = _get_auth_headers()
    auth_header = headers.get("Authorization", "")

    if auth_header.startswith("Bearer "):
        if BEARER_TOKEN:
            log(f"  {Colors.GREEN}Auth: Bearer (MCPGATEWAY_BEARER_TOKEN){Colors.RESET}")
        else:
            log(f"  {Colors.GREEN}Auth: Bearer (auto-generated JWT){Colors.RESET}")
    elif auth_header.startswith("Basic "):
        log(f"  {Colors.YELLOW}Auth: Basic (WARNING: /rpc calls may fail){Colors.RESET}")
    else:
        log(f"  {Colors.RED}Auth: None (WARNING: /rpc calls will fail){Colors.RESET}")


def _json_rpc_request(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Create a JSON-RPC 2.0 request."""
    return {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": method,
        "params": params or {},
    }


def _fetch_json(url: str, headers: dict[str, str], timeout: float = 30.0) -> tuple[int, Any]:
    """Fetch JSON from URL using urllib (gevent-safe).

    Args:
        url: Full URL to fetch
        headers: HTTP headers to include
        timeout: Request timeout in seconds

    Returns:
        Tuple of (status_code, json_data or None)
    """
    import json  # pylint: disable=import-outside-toplevel
    import urllib.error  # pylint: disable=import-outside-toplevel
    import urllib.request  # pylint: disable=import-outside-toplevel

    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            return (resp.status, data)
    except urllib.error.HTTPError as e:
        return (e.code, None)
    except Exception:
        return (0, None)


# =============================================================================
# User Classes - Full-featured (matching load-test-ui)
# =============================================================================


class BaseUser(FastHttpUser):
    """Base user class with common configuration.

    Uses FastHttpUser (gevent-based) for maximum throughput.
    Optimized for 10000+ concurrent users with aggressive timing.
    """

    abstract = True
    wait_time = between(0.05, 0.2)  # Slightly relaxed to reduce connection pressure

    # Connection settings optimized for high load
    connection_timeout = 10.0  # Allow more time for connection under load
    network_timeout = 15.0  # Allow more time for response under load

    # Increase connection pool to handle high concurrency
    pool_manager_class = None  # Use default pool manager
    concurrency = 10  # Max concurrent requests per user

    # Retry settings to handle transient failures (HTTP parse errors, etc.)
    insecure = False

    def __init__(self, *args, **kwargs):
        """Initialize base user with auth headers."""
        super().__init__(*args, **kwargs)
        self.auth_headers: dict[str, str] = {}
        self.admin_headers: dict[str, str] = {}

    def on_start(self):
        """Set up authentication for the user."""
        self.auth_headers = _get_auth_headers()
        self.admin_headers = {
            **self.auth_headers,
            "Accept": "text/html",
        }

    def _validate_json_response(self, response, allowed_codes: list[int] | None = None):
        """Validate response is successful and contains valid JSON."""
        allowed = allowed_codes or [200]
        if response.status_code not in allowed:
            response.failure(f"Expected {allowed}, got {response.status_code}")
            return False
        try:
            data = response.json()
            if data is None:
                response.failure("Response JSON is null")
                return False
        except Exception as e:
            response.failure(f"Invalid JSON: {e}")
            return False
        response.success()
        return True

    def _validate_html_response(self, response, allowed_codes: list[int] | None = None):
        """Validate response is successful HTML."""
        allowed = allowed_codes or [200]
        if response.status_code not in allowed:
            response.failure(f"Expected {allowed}, got {response.status_code}")
            return False
        content_type = response.headers.get("content-type", "")
        if "text/html" not in content_type:
            response.failure(f"Expected HTML, got {content_type}")
            return False
        response.success()
        return True

    def _validate_status(self, response, allowed_codes: list[int] | None = None):
        """Validate response status code only."""
        allowed = allowed_codes or [200]
        if response.status_code not in allowed:
            response.failure(f"Expected {allowed}, got {response.status_code}")
            return False
        response.success()
        return True

    def _validate_jsonrpc_response(self, response, allowed_codes: list[int] | None = None):
        """Validate response is successful JSON-RPC (no error field)."""
        allowed = allowed_codes or [200]
        if response.status_code not in allowed:
            response.failure(f"Expected {allowed}, got {response.status_code}")
            return False
        try:
            data = response.json()
            if data is None:
                response.failure("Response JSON is null")
                return False
            if "error" in data:
                error_obj = data["error"]
                error_code = error_obj.get("code", "unknown")
                error_msg = error_obj.get("message", "Unknown error")
                error_data = str(error_obj.get("data", ""))[:100]
                response.failure(f"JSON-RPC error [{error_code}]: {error_msg} - {error_data}")
                return False
        except Exception as e:
            response.failure(f"Invalid JSON: {e}")
            return False
        response.success()
        return True


class HealthCheckUser(BaseUser):
    """User that only performs health checks."""

    weight = 1
    wait_time = between(0.1, 0.3)  # Fast health checks

    @task(10)
    @tag("health", "critical")
    def health_check(self):
        """Check the health endpoint (no auth required)."""
        try:
            with self.client.get("/health", name="/health", catch_response=True) as response:
                self._validate_status(response)
        except Exception:
            pass  # Connection errors are recorded via catch_response

    @task(5)
    @tag("health")
    def readiness_check(self):
        """Check readiness endpoint (no auth required)."""
        try:
            with self.client.get("/ready", name="/ready", catch_response=True) as response:
                self._validate_status(response)
        except Exception:
            pass  # Connection errors are recorded via catch_response

    @task(2)
    @tag("health")
    def metrics_endpoint(self):
        """Check Prometheus metrics endpoint."""
        try:
            with self.client.get("/metrics", headers=self.auth_headers, name="/metrics", catch_response=True) as response:
                self._validate_status(response)
        except Exception:
            pass  # Connection errors are recorded via catch_response


class ReadOnlyAPIUser(BaseUser):
    """User that performs read-only API operations."""

    weight = 5
    wait_time = between(0.01, 0.1)  # Aggressive

    @task(10)
    @tag("api", "tools")
    def list_tools(self):
        """List all tools."""
        try:
            with self.client.get("/tools", headers=self.auth_headers, name="/tools", catch_response=True) as response:
                self._validate_json_response(response)
        except Exception:
            pass

    @task(8)
    @tag("api", "servers")
    def list_servers(self):
        """List all servers."""
        try:
            with self.client.get("/servers", headers=self.auth_headers, name="/servers", catch_response=True) as response:
                self._validate_json_response(response)
        except Exception:
            pass

    @task(6)
    @tag("api", "gateways")
    def list_gateways(self):
        """List all gateways."""
        try:
            with self.client.get("/gateways", headers=self.auth_headers, name="/gateways", catch_response=True) as response:
                self._validate_json_response(response)
        except Exception:
            pass

    @task(5)
    @tag("api", "resources")
    def list_resources(self):
        """List all resources."""
        try:
            with self.client.get("/resources", headers=self.auth_headers, name="/resources", catch_response=True) as response:
                self._validate_json_response(response)
        except Exception:
            pass

    @task(5)
    @tag("api", "prompts")
    def list_prompts(self):
        """List all prompts."""
        try:
            with self.client.get("/prompts", headers=self.auth_headers, name="/prompts", catch_response=True) as response:
                self._validate_json_response(response)
        except Exception:
            pass

    @task(3)
    @tag("api", "tools")
    def get_single_tool(self):
        """Get a specific tool by ID."""
        if TOOL_IDS:
            try:
                tool_id = random.choice(TOOL_IDS)
                with self.client.get(f"/tools/{tool_id}", headers=self.auth_headers, name="/tools/[id]", catch_response=True) as response:
                    self._validate_json_response(response, allowed_codes=[200, 404])
            except Exception:
                pass

    @task(3)
    @tag("api", "servers")
    def get_single_server(self):
        """Get a specific server by ID."""
        if SERVER_IDS:
            try:
                server_id = random.choice(SERVER_IDS)
                with self.client.get(f"/servers/{server_id}", headers=self.auth_headers, name="/servers/[id]", catch_response=True) as response:
                    self._validate_json_response(response, allowed_codes=[200, 404])
            except Exception:
                pass


class MCPJsonRpcUser(BaseUser):
    """User that makes MCP JSON-RPC requests."""

    weight = 4
    wait_time = between(0.01, 0.1)  # Aggressive

    def _rpc_request(self, payload: dict, name: str):
        """Make an RPC request with proper error handling."""
        try:
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name=name,
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass  # Connection errors are expected during stress testing

    @task(10)
    @tag("mcp", "rpc", "tools")
    def rpc_list_tools(self):
        """JSON-RPC: List tools."""
        payload = _json_rpc_request("tools/list")
        self._rpc_request(payload, "/rpc tools/list")

    @task(8)
    @tag("mcp", "rpc", "resources")
    def rpc_list_resources(self):
        """JSON-RPC: List resources."""
        payload = _json_rpc_request("resources/list")
        self._rpc_request(payload, "/rpc resources/list")

    @task(8)
    @tag("mcp", "rpc", "prompts")
    def rpc_list_prompts(self):
        """JSON-RPC: List prompts."""
        payload = _json_rpc_request("prompts/list")
        self._rpc_request(payload, "/rpc prompts/list")

    @task(5)
    @tag("mcp", "rpc", "tools")
    def rpc_call_tool(self):
        """JSON-RPC: Call a tool with empty arguments."""
        callable_tools = [t for t in TOOL_NAMES if t not in TOOLS_WITH_REQUIRED_ARGS]
        if callable_tools:
            tool_name = random.choice(callable_tools)
            payload = _json_rpc_request("tools/call", {"name": tool_name, "arguments": {}})
            self._rpc_request(payload, "/rpc tools/call")

    @task(4)
    @tag("mcp", "rpc", "resources")
    def rpc_read_resource(self):
        """JSON-RPC: Read a resource."""
        if RESOURCE_URIS:
            resource_uri = random.choice(RESOURCE_URIS)
            payload = _json_rpc_request("resources/read", {"uri": resource_uri})
            self._rpc_request(payload, "/rpc resources/read")

    @task(3)
    @tag("mcp", "rpc", "initialize")
    def rpc_initialize(self):
        """JSON-RPC: Initialize session."""
        payload = _json_rpc_request(
            "initialize",
            {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
                "clientInfo": {"name": "spin-detector", "version": "1.0.0"},
            },
        )
        self._rpc_request(payload, "/rpc initialize")

    @task(2)
    @tag("mcp", "rpc", "ping")
    def rpc_ping(self):
        """JSON-RPC: Ping."""
        payload = _json_rpc_request("ping")
        self._rpc_request(payload, "/rpc ping")


class FastTimeUser(BaseUser):
    """User that calls the fast_time MCP server tools."""

    weight = 5
    wait_time = between(0.01, 0.1)  # Aggressive

    def _rpc_request(self, payload: dict, name: str):
        """Make an RPC request with proper error handling."""
        try:
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name=name,
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass  # Connection errors are expected during stress testing

    @task(10)
    @tag("mcp", "fasttime", "tools")
    def call_get_system_time(self):
        """Call fast-time-get-system-time with Europe/Dublin timezone."""
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-time-get-system-time",
                "arguments": {"timezone": "Europe/Dublin"},
            },
        )
        self._rpc_request(payload, "/rpc fast-time-get-system-time")

    @task(5)
    @tag("mcp", "fasttime", "tools")
    def call_get_system_time_utc(self):
        """Call fast-time-get-system-time with UTC timezone."""
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-time-get-system-time",
                "arguments": {"timezone": "UTC"},
            },
        )
        self._rpc_request(payload, "/rpc fast-time-get-system-time [UTC]")

    @task(3)
    @tag("mcp", "fasttime", "tools")
    def call_convert_time(self):
        """Call fast-time-convert-time to convert between timezones."""
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-time-convert-time",
                "arguments": {
                    "time": "2025-01-01T12:00:00",
                    "source_timezone": "UTC",
                    "target_timezone": "Europe/Dublin",
                },
            },
        )
        self._rpc_request(payload, "/rpc fast-time-convert-time")


class FastTestEchoUser(BaseUser):
    """User that calls the fast_test MCP server echo tool."""

    weight = 3
    wait_time = between(0.01, 0.1)  # Aggressive

    ECHO_MESSAGES = [
        "Hello, World!",
        "Testing MCP protocol",
        "Load test in progress",
        "Performance benchmark",
        "Echo echo echo",
        "The quick brown fox jumps over the lazy dog",
    ]

    def _rpc_request(self, payload: dict, name: str):
        """Make an RPC request with proper error handling."""
        try:
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name=name,
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass  # Connection errors are expected during stress testing

    @task(10)
    @tag("mcp", "fasttest", "echo")
    def call_echo(self):
        """Call fast-test-echo with a random message."""
        message = random.choice(self.ECHO_MESSAGES)
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-test-echo",
                "arguments": {"message": message},
            },
        )
        self._rpc_request(payload, "/rpc fast-test-echo")

    @task(5)
    @tag("mcp", "fasttest", "echo")
    def call_echo_short(self):
        """Call fast-test-echo with a short message."""
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-test-echo",
                "arguments": {"message": "ping"},
            },
        )
        self._rpc_request(payload, "/rpc fast-test-echo [short]")


class FastTestTimeUser(BaseUser):
    """User that calls the fast_test MCP server get_system_time tool."""

    weight = 3
    wait_time = between(0.01, 0.1)  # Aggressive

    TIMEZONES = [
        "UTC",
        "America/New_York",
        "America/Los_Angeles",
        "Europe/London",
        "Europe/Paris",
        "Europe/Dublin",
        "Asia/Tokyo",
    ]

    def _rpc_request(self, payload: dict, name: str):
        """Make an RPC request with proper error handling."""
        try:
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name=name,
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass  # Connection errors are expected during stress testing

    @task(10)
    @tag("mcp", "fasttest", "time")
    def call_get_system_time(self):
        """Call fast-time-get-system-time with a random timezone."""
        timezone = random.choice(self.TIMEZONES)
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-test-get-system-time",
                "arguments": {"timezone": timezone},
            },
        )
        self._rpc_request(payload, "/rpc fast-test-get-system-time")

    @task(2)
    @tag("mcp", "fasttest", "stats")
    def call_get_stats(self):
        """Call fast-test-get-stats to get server statistics."""
        payload = _json_rpc_request(
            "tools/call",
            {
                "name": "fast-test-get-stats",
                "arguments": {},
            },
        )
        self._rpc_request(payload, "/rpc fast-test-get-stats")


class WriteAPIUser(BaseUser):
    """User that performs write operations."""

    weight = 1
    wait_time = between(0.1, 0.3)  # Faster writes for stress

    def __init__(self, *args, **kwargs):
        """Initialize with tracking for cleanup."""
        super().__init__(*args, **kwargs)
        self.created_tools: list[str] = []

    def on_stop(self):
        """Clean up created entities."""
        for tool_id in self.created_tools:
            try:
                self.client.delete(f"/tools/{tool_id}", headers=self.auth_headers, name="/tools/[id] [cleanup]")
            except Exception:
                pass

    @task(5)
    @tag("api", "write", "tools")
    def create_and_delete_tool(self):
        """Create a tool and then delete it."""
        try:
            tool_name = f"spintest-tool-{uuid.uuid4().hex[:8]}"
            tool_data = {
                "name": tool_name,
                "description": "Spin detector test tool - will be deleted",
                "integration_type": "MCP",
                "input_schema": {"type": "object", "properties": {"input": {"type": "string"}}},
            }

            with self.client.post(
                "/tools",
                json=tool_data,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name="/tools [create]",
                catch_response=True,
            ) as response:
                if response.status_code in (200, 201):
                    try:
                        data = response.json()
                        tool_id = data.get("id") or data.get("name") or tool_name
                        time.sleep(0.1)
                        self.client.delete(f"/tools/{tool_id}", headers=self.auth_headers, name="/tools/[id] [delete]")
                    except Exception:
                        pass
                elif response.status_code in (409, 422):
                    response.success()
        except Exception:
            pass  # Connection errors are expected during stress testing


class StressTestUser(BaseUser):
    """User for stress testing with predictable request rate."""

    weight = 1
    wait_time = constant_throughput(2)

    @task(10)
    @tag("stress", "health")
    def rapid_health_check(self):
        """Rapid health checks."""
        try:
            self.client.get("/health", name="/health [stress]")
        except Exception:
            pass

    @task(8)
    @tag("stress", "api")
    def rapid_tools_list(self):
        """Rapid tools listing."""
        try:
            self.client.get("/tools", headers=self.auth_headers, name="/tools [stress]")
        except Exception:
            pass

    @task(5)
    @tag("stress", "rpc")
    def rapid_rpc_ping(self):
        """Rapid RPC pings."""
        try:
            payload = _json_rpc_request("ping")
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name="/rpc ping [stress]",
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass


class RealisticUser(BaseUser):
    """User that simulates realistic mixed traffic."""

    weight = 10
    wait_time = between(0.01, 0.1)  # Aggressive for stress testing

    @task(15)
    @tag("realistic", "health")
    def health_check(self):
        """Health check."""
        try:
            self.client.get("/health", name="/health")
        except Exception:
            pass

    @task(20)
    @tag("realistic", "api")
    def list_tools(self):
        """List tools."""
        try:
            self.client.get("/tools", headers=self.auth_headers, name="/tools")
        except Exception:
            pass

    @task(15)
    @tag("realistic", "api")
    def list_servers(self):
        """List servers."""
        try:
            self.client.get("/servers", headers=self.auth_headers, name="/servers")
        except Exception:
            pass

    @task(10)
    @tag("realistic", "api")
    def list_gateways(self):
        """List gateways."""
        try:
            self.client.get("/gateways", headers=self.auth_headers, name="/gateways")
        except Exception:
            pass

    @task(10)
    @tag("realistic", "rpc")
    def rpc_list_tools(self):
        """JSON-RPC list tools."""
        try:
            payload = _json_rpc_request("tools/list")
            with self.client.post(
                "/rpc",
                json=payload,
                headers={**self.auth_headers, "Content-Type": "application/json"},
                name="/rpc tools/list",
                catch_response=True,
            ) as response:
                self._validate_jsonrpc_response(response)
        except Exception:
            pass

    @task(5)
    @tag("realistic", "api")
    def get_single_tool(self):
        """Get specific tool."""
        if TOOL_IDS:
            try:
                tool_id = random.choice(TOOL_IDS)
                with self.client.get(f"/tools/{tool_id}", headers=self.auth_headers, name="/tools/[id]", catch_response=True) as response:
                    self._validate_json_response(response, allowed_codes=[200, 404])
            except Exception:
                pass


# =============================================================================
# Event Handlers
# =============================================================================


@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    """Initialize test environment."""
    # Suppress noisy Locust runner logs unless LOCUST_VERBOSE=1
    # This must be done here because Locust configures logging after module import
    if not VERBOSE_LOGGING:
        logging.getLogger("locust.runners").setLevel(logging.WARNING)
        logging.getLogger("locust.main").setLevel(logging.WARNING)
        logging.getLogger("locust").setLevel(logging.WARNING)

    # Debug logging for runner type
    if isinstance(environment.runner, MasterRunner):
        logger.debug("Running as master node")
    elif isinstance(environment.runner, WorkerRunner):
        logger.debug("Running as worker node")
    else:
        logger.debug("Running in standalone mode")


@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **_kwargs):
    """Handle request events, suppressing noisy error logging for expected failures."""
    # HTTP parse errors and connection errors are expected during extreme load
    # Don't log them individually as they're already counted in stats
    if exception:
        error_msg = str(exception)
        if "Expected HTTP/" in error_msg or "Connection reset" in error_msg:
            # Silently count these as they're expected during stress testing
            pass


@events.test_start.add_listener
def on_test_start(environment, **_kwargs):
    """Fetch existing entity IDs for use in tests (master/standalone only)."""
    # Only fetch on master or standalone - workers don't need this
    if isinstance(environment.runner, WorkerRunner):
        return

    host = environment.host or "http://localhost:8080"
    headers = _get_auth_headers()

    try:
        # Fetch tools
        status, data = _fetch_json(f"{host}/tools", headers)
        if status == 200 and data:
            items = data if isinstance(data, list) else data.get("tools", data.get("items", []))
            TOOL_IDS.extend([str(t.get("id")) for t in items[:50] if t.get("id")])
            TOOL_NAMES.extend([str(t.get("name")) for t in items[:50] if t.get("name")])

        # Fetch servers
        status, data = _fetch_json(f"{host}/servers", headers)
        if status == 200 and data:
            items = data if isinstance(data, list) else data.get("servers", data.get("items", []))
            SERVER_IDS.extend([str(s.get("id")) for s in items[:50] if s.get("id")])

        # Fetch gateways
        status, data = _fetch_json(f"{host}/gateways", headers)
        if status == 200 and data:
            items = data if isinstance(data, list) else data.get("gateways", data.get("items", []))
            GATEWAY_IDS.extend([str(g.get("id")) for g in items[:50] if g.get("id")])

        # Fetch resources
        status, data = _fetch_json(f"{host}/resources", headers)
        if status == 200 and data:
            items = data if isinstance(data, list) else data.get("resources", data.get("items", []))
            RESOURCE_IDS.extend([str(r.get("id")) for r in items[:50] if r.get("id")])
            RESOURCE_URIS.extend([str(r.get("uri")) for r in items[:50] if r.get("uri")])

        # Fetch prompts
        status, data = _fetch_json(f"{host}/prompts", headers)
        if status == 200 and data:
            items = data if isinstance(data, list) else data.get("prompts", data.get("items", []))
            PROMPT_IDS.extend([str(p.get("id")) for p in items[:50] if p.get("id")])
            PROMPT_NAMES.extend([str(p.get("name")) for p in items[:50] if p.get("name")])

    except Exception:
        pass  # Tests will continue without pre-fetched IDs


@events.test_stop.add_listener
def on_test_stop(environment, **_kwargs):
    """Clean up on test stop (master/standalone only)."""
    # Only run cleanup on master or standalone
    if isinstance(environment.runner, WorkerRunner):
        return

    TOOL_IDS.clear()
    SERVER_IDS.clear()
    GATEWAY_IDS.clear()
    RESOURCE_IDS.clear()
    PROMPT_IDS.clear()
    TOOL_NAMES.clear()
    RESOURCE_URIS.clear()
    PROMPT_NAMES.clear()

    _close_log_file()
    log(f"\n{Colors.DIM}Log saved to: {LOG_FILE}{Colors.RESET}\n")


# Stats tracking for RPS display
_last_total_requests = 0
_last_stats_time = 0.0
_stats_interval = 5  # Print stats every 5 seconds


@events.report_to_master.add_listener
def on_report_to_master(client_id, data):
    """Hook for worker reporting (no-op, stats are aggregated on master)."""
    pass


def _print_rps_stats(environment, force: bool = False) -> None:
    """Print current RPS and user count.

    Args:
        environment: Locust environment with stats and runner.
        force: If True, print even if stats interval hasn't elapsed.
    """
    global _last_total_requests, _last_stats_time

    if not environment:
        return

    stats = environment.stats
    current_time = time.time()
    runner = environment.runner

    if _last_stats_time == 0:
        _last_stats_time = current_time
        _last_total_requests = stats.total.num_requests
        if not force:
            return

    elapsed = current_time - _last_stats_time
    if elapsed < _stats_interval and not force:
        return

    current_requests = stats.total.num_requests
    delta_requests = current_requests - _last_total_requests
    rps = delta_requests / elapsed if elapsed > 0 else 0

    user_count = runner.user_count if runner else 0
    fail_count = stats.total.num_failures
    fail_ratio = stats.total.fail_ratio * 100 if stats.total.num_requests > 0 else 0

    # Format RPS display with color coding
    if rps > 1000:
        rps_color = Colors.GREEN
    elif rps > 500:
        rps_color = Colors.CYAN
    elif rps > 100:
        rps_color = Colors.YELLOW
    else:
        rps_color = Colors.DIM

    # Calculate avg response time
    avg_response = stats.total.avg_response_time

    # Build status line with RPS prominently displayed
    status = (
        f"  📊 {Colors.BOLD}RPS:{Colors.RESET} {rps_color}{rps:>6,.0f}{Colors.RESET} | "
        f"{Colors.BOLD}Users:{Colors.RESET} {user_count:>5,} | "
        f"{Colors.BOLD}Avg:{Colors.RESET} {avg_response:>6.0f}ms | "
        f"{Colors.BOLD}Fail:{Colors.RESET} "
    )

    if fail_ratio > 10:
        status += f"{Colors.RED}{fail_count:,} ({fail_ratio:.1f}%){Colors.RESET}"
    elif fail_ratio > 1:
        status += f"{Colors.YELLOW}{fail_count:,} ({fail_ratio:.1f}%){Colors.RESET}"
    else:
        status += f"{Colors.GREEN}{fail_count:,} ({fail_ratio:.1f}%){Colors.RESET}"

    log(status)

    _last_total_requests = current_requests
    _last_stats_time = current_time


# =============================================================================
# Load Shape - Spike/Drop Pattern for CPU Spin Detection
# =============================================================================
class SpinDetectorShape(LoadTestShape):
    """Load shape with spike/drop pattern for detecting CPU spin loops.

    Pattern:
    - Ramp up to target users over ramp_time
    - Sustain load for sustain_time
    - Drop to 0 users
    - Pause for pause_time (monitor CPU - should return to idle)
    - Repeat for multiple cycles

    If CPU stays high during pause phases, the spin loop bug is present.
    """

    # Configuration for each cycle: (target_users, sustain_time, pause_time)
    # ESCALATING pattern: progressively more users and longer load phases
    # Format: (users, sustain_time, pause_time)
    # Ramp time is calculated automatically based on spawn_rate
    #
    # Pattern: 4K → 6K → 8K → 10K → 10K (repeat forever)
    cycles = [
        (4000, 30, 10),  # Wave 1: 4K users, 30s sustain, 10s pause
        (6000, 45, 15),  # Wave 2: 6K users, 45s sustain, 15s pause
        (8000, 60, 20),  # Wave 3: 8K users, 60s sustain, 20s pause
        (10000, 75, 30),  # Wave 4: 10K users, 75s sustain, 30s pause
        (10000, 90, 30),  # Wave 5: 10K users, 90s sustain, 30s pause
    ]

    spawn_rate = 1000  # Always 1000/s spawn rate

    def __init__(self):
        """Initialize the load shape."""
        super().__init__()
        self._current_cycle = 0
        self._cycle_start_time = 0
        self._last_phase = None
        self._pause_stats: list[tuple[int, float]] = []  # (cycle, max_cpu) during pauses
        self._banner_printed = False
        self._total_cycles = 0  # Track total cycles across all iterations
        self._last_rps_print = 0.0  # Last time RPS was printed

    def tick(self) -> Optional[tuple[int, float]]:
        """Calculate the current target user count and spawn rate."""
        # Print banner on first tick (before any phase output)
        if not self._banner_printed:
            self._banner_printed = True
            self._print_banner()

        run_time = self.get_run_time()

        # Loop indefinitely - reset to cycle 0 when we reach the end
        if self._current_cycle >= len(self.cycles):
            self._current_cycle = 0
            self._last_phase = None  # Reset phase to trigger new cycle logging

        target_users, sustain_time, pause_time = self.cycles[self._current_cycle]
        # Calculate ramp time based on spawn rate (e.g., 4000 users / 1000 spawn = 4 seconds)
        ramp_time = max(1, target_users // self.spawn_rate)
        cycle_duration = ramp_time + sustain_time + pause_time

        if self._cycle_start_time == 0:
            self._cycle_start_time = run_time

        cycle_time = run_time - self._cycle_start_time

        if cycle_time < ramp_time:
            phase = "ramp"
            progress = cycle_time / ramp_time
            users = max(1, int(target_users * progress))  # At least 1 user during ramp
        elif cycle_time < ramp_time + sustain_time:
            phase = "sustain"
            users = target_users
        elif cycle_time < cycle_duration:
            phase = "pause"
            users = 0
        else:
            self._current_cycle += 1
            self._total_cycles += 1
            self._cycle_start_time = run_time
            self._last_phase = None
            return self.tick()

        if phase != self._last_phase:
            self._log_phase_change(phase, users, target_users)
            self._last_phase = phase

        # Print RPS stats every 5 seconds during load phases
        current_time = time.time()
        if phase in ("ramp", "sustain") and current_time - self._last_rps_print >= 5:
            if hasattr(self, "runner") and self.runner and hasattr(self.runner, "environment"):
                _print_rps_stats(self.runner.environment)
            self._last_rps_print = current_time

        return (users, self.spawn_rate)

    def _print_banner(self) -> None:
        """Print initial banner and instructions."""
        _init_log_file()

        log("")
        print_box(
            "CPU SPIN LOOP DETECTOR",
            f"Issue #2360 | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            Colors.CYAN,
        )

        log(f"""
{Colors.BOLD}PURPOSE:{Colors.RESET}
  Detect CPU spin loop bug caused by orphaned asyncio tasks.

{Colors.BOLD}ESCALATING LOAD PATTERN (1000/s spawn rate):{Colors.RESET}
  {Colors.GREEN}Wave 1:{Colors.RESET}  4,000 users for 30s → {Colors.MAGENTA}10s pause{Colors.RESET}
  {Colors.GREEN}Wave 2:{Colors.RESET}  6,000 users for 45s → {Colors.MAGENTA}15s pause{Colors.RESET}
  {Colors.YELLOW}Wave 3:{Colors.RESET}  8,000 users for 60s → {Colors.MAGENTA}20s pause{Colors.RESET}
  {Colors.RED}Wave 4:{Colors.RESET} 10,000 users for 75s → {Colors.MAGENTA}30s pause{Colors.RESET}
  {Colors.RED}Wave 5:{Colors.RESET} 10,000 users for 90s → {Colors.MAGENTA}30s pause{Colors.RESET}
  → {Colors.BOLD}Repeat forever (Ctrl+C to stop){Colors.RESET}

{Colors.BOLD}EXPECTED:{Colors.RESET}
  {Colors.GREEN}PASS:{Colors.RESET} CPU <10% during pause | {Colors.RED}FAIL:{Colors.RESET} CPU >100% during pause

{Colors.BOLD}MONITORING:{Colors.RESET}
  Log file: {Colors.CYAN}{LOG_FILE}{Colors.RESET}
  Monitor:  {Colors.DIM}tail -f {LOG_FILE}{Colors.RESET}

{Colors.BOLD}CONFIGURATION:{Colors.RESET}
  Workers:  {Colors.CYAN}{WORKER_COUNT}{Colors.RESET} (processes spawning users)
""")

        _log_auth_mode()

        print_section("Initial Docker Stats")
        stats_output, cpu_values = get_docker_stats()
        log(stats_output)
        log(f"\n{format_cpu_status(cpu_values)}\n")

    def _log_phase_change(self, phase: str, current_users: int, target_users: int) -> None:
        """Log phase transitions with docker stats."""
        # Use total cycles for display (1-indexed)
        cycle_num = self._total_cycles + 1
        cycle_letter = chr(ord("A") + (self._current_cycle % len(self.cycles)))

        stats_output, cpu_values = get_docker_stats()
        # Only flag high CPU as a problem during pause phases (spin loop detection)
        is_pause = phase == "pause"
        cpu_status = format_cpu_status(cpu_values, is_pause_phase=is_pause)

        if phase == "ramp":
            print_box(
                f"CYCLE {cycle_num} ({cycle_letter}): RAMPING UP",
                f"Target: {target_users} users | Spawn rate: {self.spawn_rate}/s",
                Colors.BLUE,
            )
        elif phase == "sustain":
            print_box(
                f"CYCLE {cycle_num} ({cycle_letter}): SUSTAINING LOAD",
                f"Holding at {target_users} users",
                Colors.CYAN,
            )
        elif phase == "pause":
            print_box(
                f"CYCLE {cycle_num} ({cycle_letter}): PAUSE - MONITORING CPU",
                "",
                Colors.MAGENTA,
            )
            log("")
            log(f"  {Colors.YELLOW}{Colors.BOLD}>>> ALL USERS DISCONNECTED <<<{Colors.RESET}")
            log(f"  {Colors.YELLOW}>>> CPU should drop to <10% if cleanup is working <<<{Colors.RESET}")
            log("")

            # Record max CPU for this pause
            if cpu_values:
                max_cpu = max(cpu for _, cpu in cpu_values)
                self._pause_stats.append((cycle_num, max_cpu))

                # Log summary every 5 cycles
                if cycle_num % 5 == 0:
                    self._log_periodic_summary()

        # Print docker stats
        print_section("Docker Stats")
        log(stats_output)
        log(f"\n{cpu_status}\n")

        # Print RPS stats if we have access to the runner (force print on phase change)
        if hasattr(self, "runner") and self.runner and hasattr(self.runner, "environment"):
            print_section("Request Stats")
            _print_rps_stats(self.runner.environment, force=True)

    def _log_periodic_summary(self) -> None:
        """Log a periodic summary of pause phase CPU stats."""
        if not self._pause_stats:
            return

        # Get last 5 cycles
        recent = self._pause_stats[-5:]
        passes = sum(1 for _, cpu in recent if cpu < 10)
        warns = sum(1 for _, cpu in recent if 10 <= cpu < 50)
        fails = sum(1 for _, cpu in recent if cpu >= 50)

        log("")
        log(f"{Colors.CYAN}{Colors.BOLD}=== PERIODIC SUMMARY (last 5 cycles) ==={Colors.RESET}")
        log(f"  {Colors.GREEN}PASS: {passes}{Colors.RESET} | {Colors.YELLOW}WARN: {warns}{Colors.RESET} | {Colors.RED}FAIL: {fails}{Colors.RESET}")
        log(f"  Total cycles completed: {len(self._pause_stats)}")
        log("")

    def _print_final_report(self) -> None:
        """Print final summary report."""
        stats_output, cpu_values = get_docker_stats()

        log("")
        print_box(
            "TEST COMPLETE",
            "",
            Colors.GREEN if all(cpu < 10 for _, cpu in cpu_values) else Colors.RED,
        )

        # Summary table
        print_section("Pause Phase CPU Summary", Colors.MAGENTA)
        log(f"{'Cycle':<10} {'Max CPU %':<15} {'Status':<20}")
        log("-" * 45)

        all_passed = True
        for cycle_num, max_cpu in self._pause_stats:
            if max_cpu < 10:
                status = f"{Colors.GREEN}PASS{Colors.RESET}"
            elif max_cpu < 50:
                status = f"{Colors.YELLOW}WARN{Colors.RESET}"
                all_passed = False
            else:
                status = f"{Colors.RED}FAIL{Colors.RESET}"
                all_passed = False
            log(f"{cycle_num:<10} {max_cpu:<15.1f} {status}")

        # Final stats (check as pause phase - CPU should be idle)
        print_section("Final Docker Stats")
        log(stats_output)
        log(f"\n{format_cpu_status(cpu_values, is_pause_phase=True)}")

        # Verdict
        log("")
        if all_passed and cpu_values and all(cpu < 10 for _, cpu in cpu_values):
            log(f"{Colors.GREEN}{Colors.BOLD}")
            log("  +------------------------------------------+")
            log("  |              TEST PASSED                 |")
            log("  |   CPU returned to idle after cleanup     |")
            log("  +------------------------------------------+")
            log(f"{Colors.RESET}")
        else:
            log(f"{Colors.RED}{Colors.BOLD}")
            log("  +------------------------------------------+")
            log("  |              TEST FAILED                 |")
            log("  |   CPU spin loop may still be present     |")
            log("  +------------------------------------------+")
            log(f"{Colors.RESET}")
            log(f"\n{Colors.YELLOW}See: todo/how-to-analyze.md for debugging steps{Colors.RESET}")

        log(f"\n{Colors.DIM}Issue: https://github.com/IBM/mcp-context-forge/issues/2360{Colors.RESET}")
        log(f"{Colors.DIM}Log file: {LOG_FILE}{Colors.RESET}\n")
