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

Concurrency tests for row-level locking implementation.
Tests verify that concurrent operations on tools, gateways, prompts, resources,
A2A agents, and servers handle race conditions correctly using PostgreSQL row-level locking.

This test suite validates:
1. Tool creation with duplicate names (public and team visibility)
2. Tool updates with name conflicts
3. Tool toggle operations under concurrent load
4. Gateway creation with duplicate slugs (public and team visibility)
5. Gateway updates with slug conflicts
6. Prompt creation with duplicate names and concurrent operations
7. Resource creation with duplicate URIs and concurrent operations
8. A2A agent creation with duplicate names and concurrent operations
9. Server creation with duplicate names and concurrent operations
10. Mixed concurrent operations (create, update, toggle, read, delete)
11. High concurrency scenarios with unique entities
12. Row-level locking with skip_locked behavior
13. Mixed visibility operations (public, team, private)
"""

# Standard
import asyncio
import os
import uuid

# Third-Party
from httpx import AsyncClient
import pytest
import pytest_asyncio

# First-Party
from mcpgateway.db import get_db

# Set environment variables for testing
os.environ["MCPGATEWAY_ADMIN_API_ENABLED"] = "true"
os.environ["MCPGATEWAY_UI_ENABLED"] = "true"
os.environ["MCPGATEWAY_A2A_ENABLED"] = "true"


def is_postgresql() -> bool:
    """Check if PostgreSQL is configured via DB env var or DATABASE_URL."""
    db_env = os.getenv("DB", "").lower()
    database_url = os.getenv("DATABASE_URL", "").lower()
    return db_env == "postgres" or "postgresql" in database_url


# Skip condition for PostgreSQL-only tests
SKIP_IF_NOT_POSTGRES = pytest.mark.skipif(not is_postgresql(), reason="Row-level locking only works on PostgreSQL")


def create_test_jwt_token():
    """Create a proper JWT token for testing."""
    import datetime
    import jwt

    test_secret = "integration-test-jwt-secret-key-with-minimum-32-bytes"
    expire = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=60)
    payload = {
        "sub": "admin@example.com",
        "email": "admin@example.com",
        "iat": int(datetime.datetime.now(datetime.timezone.utc).timestamp()),
        "exp": int(expire.timestamp()),
        "iss": "mcpgateway",
        "aud": "mcpgateway-api",
        "teams": [],
    }
    return jwt.encode(payload, test_secret, algorithm="HS256")


TEST_JWT_TOKEN = create_test_jwt_token()
TEST_AUTH_HEADER = {"Authorization": f"Bearer {TEST_JWT_TOKEN}"}


@pytest_asyncio.fixture
async def client(app_with_temp_db):
    """Create test client with authentication mocked."""
    from mcpgateway.auth import get_current_user
    from mcpgateway.middleware.rbac import get_current_user_with_permissions
    from mcpgateway.utils.verify_credentials import require_admin_auth
    from mcpgateway.db import EmailUser
    from tests.utils.rbac_mocks import create_mock_email_user, create_mock_user_context

    TEST_USER = create_mock_email_user(email="admin@example.com", full_name="Test Admin", is_admin=True, is_active=True)

    test_db_dependency = app_with_temp_db.dependency_overrides.get(get_db) or get_db

    def get_test_db_session():
        if callable(test_db_dependency):
            return next(test_db_dependency())
        return test_db_dependency

    test_db_session = get_test_db_session()

    # Create admin user in database for permission checks (if not exists)
    from sqlalchemy import select

    existing_user = test_db_session.execute(select(EmailUser).where(EmailUser.email == "admin@example.com")).scalar_one_or_none()

    if not existing_user:
        admin_user = EmailUser(email="admin@example.com", full_name="Test Admin", is_admin=True, is_active=True, password_hash="dummy_hash")
        test_db_session.add(admin_user)
        test_db_session.commit()

    test_user_context = create_mock_user_context(email="admin@example.com", full_name="Test Admin", is_admin=True)
    test_user_context["db"] = test_db_session

    async def mock_require_admin_auth():
        return "admin@example.com"

    # Mock only the gateway initialization to prevent actual connection attempts
    # but keep the database operations intact
    from mcpgateway import admin

    original_initialize = admin.gateway_service._initialize_gateway

    async def mock_initialize_gateway(*args, **kwargs):
        # Return mock data without actually connecting
        return ({"capabilities": {}}, [], [], [])  # capabilities  # tools  # resources  # prompts

    admin.gateway_service._initialize_gateway = mock_initialize_gateway

    app_with_temp_db.dependency_overrides[get_current_user] = lambda: TEST_USER
    app_with_temp_db.dependency_overrides[get_current_user_with_permissions] = lambda: test_user_context
    app_with_temp_db.dependency_overrides[require_admin_auth] = mock_require_admin_auth

    from httpx import ASGITransport, AsyncClient

    transport = ASGITransport(app=app_with_temp_db)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    # Restore original initialization method
    admin.gateway_service._initialize_gateway = original_initialize

    app_with_temp_db.dependency_overrides.pop(get_current_user, None)
    app_with_temp_db.dependency_overrides.pop(get_current_user_with_permissions, None)
    app_with_temp_db.dependency_overrides.pop(require_admin_auth, None)


# -------------------------
# Tool Concurrency Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_tool_creation_same_name(client: AsyncClient):
    """Test concurrent tool creation with same name prevents duplicates."""
    tool_name = f"test-tool-{uuid.uuid4()}"

    async def create_tool():
        form_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Test tool", "integrationType": "REST", "requestType": "GET", "visibility": "public"}
        return await client.post("/admin/tools", data=form_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name
    results = await asyncio.gather(*[create_tool() for _ in range(10)], return_exceptions=True)

    # Count successful creations (200) and conflicts (409)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"

    # No 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [200, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_tool_update_same_name(client: AsyncClient):
    """Test concurrent tool updates to same name prevents duplicates."""
    # Create two tools
    tool1_name = f"tool-1-{uuid.uuid4()}"
    tool2_name = f"tool-2-{uuid.uuid4()}"

    tool1_data = {"name": tool1_name, "url": "http://example.com/tool1", "description": "Tool 1", "integrationType": "REST", "requestType": "GET", "visibility": "public"}
    tool2_data = {"name": tool2_name, "url": "http://example.com/tool2", "description": "Tool 2", "integrationType": "REST", "requestType": "GET", "visibility": "public"}

    resp1 = await client.post("/admin/tools", data=tool1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/admin/tools", data=tool2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 200
    assert resp2.status_code == 200

    # Get tool IDs by listing tools
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]

    tool1 = next((t for t in tools if t["name"] == tool1_name), None)
    tool2 = next((t for t in tools if t["name"] == tool2_name), None)
    assert tool1 is not None and tool2 is not None

    tool1_id = tool1["id"]
    tool2_id = tool2["id"]

    target_name = f"target-tool-{uuid.uuid4()}"

    async def update_tool(tool_id: str):
        update_data = {"name": target_name, "customName": target_name, "url": "http://example.com/updated", "requestType": "GET", "integrationType": "REST", "headers": "{}", "input_schema": "{}"}
        return await client.post(f"/admin/tools/{tool_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both tools to same name concurrently
    results = await asyncio.gather(*[update_tool(tool1_id), update_tool(tool2_id)], return_exceptions=True)

    # With row locking, one should succeed and one should fail with conflict
    # However, if both acquire locks before either commits, both may succeed
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 303])
    _conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)  # noqa: F841

    # At least one should succeed
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}. Results: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"

    # If both succeeded, verify no data corruption occurred
    if success_count == 2:
        # Both updates succeeded - verify final state is consistent
        final_list = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
        assert final_list.status_code == 200
        final_tools = final_list.json()["data"]
        # Both tools should now have the target name (last write wins)
        tools_with_target_name = [t for t in final_tools if t["name"] == target_name or t.get("customName") == target_name]
        assert len(tools_with_target_name) >= 1, "At least one tool should have the target name"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_tool_toggle(client: AsyncClient):
    """Test concurrent enable/disable doesn't cause race condition."""
    # Create a tool
    tool_name = f"toggle-tool-{uuid.uuid4()}"
    tool_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Toggle test tool", "integrationType": "REST", "requestType": "GET", "visibility": "public"}

    resp = await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get tool ID by listing tools
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]
    tool = next((t for t in tools if t["name"] == tool_name), None)
    assert tool is not None
    tool_id = tool["id"]

    async def toggle():
        return await client.post(f"/admin/tools/{tool_id}/state", data={}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should succeed or fail cleanly (no 500 errors)
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"

    # Verify final state is consistent by listing tools
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]
    final_tool = next((t for t in tools if t["id"] == tool_id), None)
    assert final_tool is not None


# -------------------------
# Gateway Concurrency Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_gateway_creation_same_slug(client: AsyncClient):
    """Test concurrent gateway creation with same slug prevents duplicates."""
    gateway_name = f"Test Gateway {uuid.uuid4()}"

    async def create_gateway():
        gateway_data = {"name": gateway_name, "url": "http://example.com/gateway", "description": "Test gateway", "visibility": "public", "transport": "SSE"}
        return await client.post("/admin/gateways", data=gateway_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name (will generate same slug)
    results = await asyncio.gather(*[create_gateway() for _ in range(10)], return_exceptions=True)

    # Count successful creations and conflicts
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    validation_error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 422)
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code >= 500)

    # Exactly one should succeed, rest should be conflicts or validation errors
    assert success_count == 1, f"Expected 1 success, got {success_count}. Status codes: {[r.status_code for r in results if not isinstance(r, Exception)]}"
    # At least 7 should be conflicts or validation errors (allowing for some timing variations)
    assert (
        conflict_count + validation_error_count
    ) >= 7, f"Expected at least 7 conflicts/validation errors, got {conflict_count} conflicts + {validation_error_count} validation errors = {conflict_count + validation_error_count}"
    assert error_count == 0, f"Expected no 500 errors, got {error_count}"

    # All non-exception responses should be either success, conflict, or validation error
    assert all(
        isinstance(r, Exception) or r.status_code in [200, 409, 422] for r in results
    ), f"Some requests returned unexpected status codes: {[r.status_code for r in results if not isinstance(r, Exception)]}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_gateway_update_same_slug(client: AsyncClient):
    """Test concurrent gateway updates to same slug prevents duplicates."""
    # Create two gateways
    gateway1_name = f"Gateway 1 {uuid.uuid4()}"
    gateway2_name = f"Gateway 2 {uuid.uuid4()}"

    gateway1_data = {"name": gateway1_name, "url": "http://example.com/gateway1", "description": "Gateway 1", "visibility": "public", "transport": "SSE"}
    gateway2_data = {"name": gateway2_name, "url": "http://example.com/gateway2", "description": "Gateway 2", "visibility": "public", "transport": "SSE"}

    resp1 = await client.post("/admin/gateways", data=gateway1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/admin/gateways", data=gateway2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 200
    assert resp2.status_code == 200

    # Get gateway IDs by listing gateways
    list_resp = await client.get("/admin/gateways", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    gateways = list_resp.json()["data"]

    gateway1 = next((g for g in gateways if g["name"] == gateway1_name), None)
    gateway2 = next((g for g in gateways if g["name"] == gateway2_name), None)
    assert gateway1 is not None and gateway2 is not None

    gateway1_id = gateway1["id"]
    gateway2_id = gateway2["id"]

    target_name = f"Target Gateway {uuid.uuid4()}"

    async def update_gateway(gateway_id: str):
        update_data = {"name": target_name, "url": "http://example.com/updated"}
        return await client.post(f"/admin/gateways/{gateway_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both gateways to same name concurrently
    results = await asyncio.gather(*[update_gateway(gateway1_id), update_gateway(gateway2_id)], return_exceptions=True)

    # Count results - in concurrent updates, we expect at least one success
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 303])
    _conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)  # noqa: F841
    _error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code >= 500)  # noqa: F841

    # At least one should succeed
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}"

    # Note: Gateway updates may encounter errors during concurrent operations
    # This is acceptable as long as at least one update succeeds
    # The test verifies that concurrent operations don't cause data corruption


# -------------------------
# Mixed Concurrency Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_mixed_operations(client: AsyncClient):
    """Test mixed concurrent operations (create, update, toggle) work correctly."""
    # Create initial tool
    tool_name = f"mixed-tool-{uuid.uuid4()}"
    tool_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Mixed test tool", "integrationType": "REST", "requestType": "GET", "visibility": "public"}

    resp = await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get tool ID by listing tools
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]
    tool = next((t for t in tools if t["name"] == tool_name), None)
    assert tool is not None
    tool_id = tool["id"]

    async def update_tool():
        update_data = {
            "name": tool_name,
            "customName": tool_name,
            "url": "http://example.com/tool",
            "description": f"Updated at {uuid.uuid4()}",
            "requestType": "GET",
            "integrationType": "REST",
            "headers": "{}",
            "input_schema": "{}",
        }
        return await client.post(f"/admin/tools/{tool_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    async def toggle_tool():
        return await client.post(f"/admin/tools/{tool_id}/state", data={}, headers=TEST_AUTH_HEADER)

    async def read_tool():
        return await client.get("/admin/tools", headers=TEST_AUTH_HEADER)

    # Mix of operations
    operations = [update_tool() for _ in range(5)] + [toggle_tool() for _ in range(5)] + [read_tool() for _ in range(5)]

    results = await asyncio.gather(*operations, return_exceptions=True)

    # All should complete without 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"

    # Verify final state is consistent
    final_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert final_resp.status_code == 200


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_high_concurrency_tool_creation(client: AsyncClient):
    """Test high concurrency with many unique tool creations."""

    async def create_unique_tool(index: int):
        tool_data = {
            "name": f"concurrent-tool-{index}-{uuid.uuid4()}",
            "url": f"http://example.com/tool{index}",
            "description": f"Concurrent test tool {index}",
            "integrationType": "REST",
            "requestType": "GET",
            "visibility": "public",
        }
        return await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)

    # Create 50 tools concurrently
    results = await asyncio.gather(*[create_unique_tool(i) for i in range(50)], return_exceptions=True)

    # All should succeed (different names)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    failed_count = sum(1 for r in results if isinstance(r, Exception) or (hasattr(r, "status_code") and r.status_code != 200))
    error_500_count = sum(1 for r in results if not isinstance(r, Exception) and hasattr(r, "status_code") and r.status_code >= 500)

    # Allow for occasional failures due to connection pool exhaustion under extreme concurrency
    # At least 48 out of 50 should succeed (96% success rate)
    assert success_count >= 48, f"Expected at least 48 successes, got {success_count} (failed: {failed_count})"

    # No 500 errors - those indicate server bugs, not resource exhaustion
    assert error_500_count == 0, f"Got {error_500_count} server errors (500+), which indicates bugs not resource limits"


# -------------------------
# Team-Scoped Tool Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_team_tool_creation_same_name(client: AsyncClient):
    """Test concurrent team tool creation with same name prevents duplicates within team."""
    tool_name = f"team-tool-{uuid.uuid4()}"

    async def create_team_tool():
        form_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Team test tool", "integrationType": "REST", "requestType": "GET", "visibility": "team"}
        return await client.post("/admin/tools", data=form_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name for team visibility
    results = await asyncio.gather(*[create_team_tool() for _ in range(10)], return_exceptions=True)

    # Count successful creations (200) and conflicts (409)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"

    # No 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [200, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_tool_update_same_tool(client: AsyncClient):
    """Test concurrent tool updates on the same tool are serialized by row locking."""
    # Create a tool with public visibility to ensure permissions work
    tool_name = f"update-test-tool-{uuid.uuid4()}"
    tool_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Update test tool", "integrationType": "REST", "requestType": "GET", "visibility": "public"}

    resp = await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get tool ID
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]
    tool = next((t for t in tools if t["name"] == tool_name), None)
    assert tool is not None
    tool_id = tool["id"]

    async def update_description():
        """Update description - row locking ensures updates are serialized."""
        update_data = {
            "name": tool_name,
            "customName": tool_name,
            "url": "http://example.com/tool",
            "description": f"Updated at {uuid.uuid4()}",
            "requestType": "GET",
            "integrationType": "REST",
            "headers": "{}",
            "input_schema": "{}",
        }
        return await client.post(f"/admin/tools/{tool_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Run concurrent updates on the same tool
    results = await asyncio.gather(*[update_description() for _ in range(5)], return_exceptions=True)

    # All updates should succeed (row locking serializes them, no conflicts on description changes)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 303])

    # All should succeed since we're just updating descriptions
    assert success_count == 5, f"Expected 5 successes, got {success_count}. Results: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"

    # No errors
    assert all(
        isinstance(r, Exception) or r.status_code in [200, 303] for r in results
    ), f"Some requests returned unexpected status codes: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_tool_delete_operations(client: AsyncClient):
    """Test that concurrent deletes of the same tool don't corrupt state.

    delete_tool uses a Core-SQL DELETE with a rowcount check: exactly one
    caller sees rowcount == 1 (success) while the others see rowcount == 0
    and raise ToolNotFoundError, which the admin endpoint surfaces as an
    error in the redirect URL.

    Note: The admin endpoint always returns 303 redirects, so we check the
    redirect URL for error messages to distinguish success from failure.
    """
    # Create a tool
    tool_name = f"delete-tool-{uuid.uuid4()}"
    tool_data = {"name": tool_name, "url": "http://example.com/tool", "description": "Delete test tool", "integrationType": "REST", "requestType": "GET", "visibility": "public"}

    resp = await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get tool ID
    list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    tools = list_resp.json()["data"]
    tool = next((t for t in tools if t["name"] == tool_name), None)
    assert tool is not None
    tool_id = tool["id"]

    async def delete_tool():
        return await client.post(f"/admin/tools/{tool_id}/delete", data={}, headers=TEST_AUTH_HEADER, follow_redirects=False)

    # Run concurrent deletes
    results = await asyncio.gather(*[delete_tool() for _ in range(5)], return_exceptions=True)

    # All should return 303 redirects (admin endpoint always redirects)
    assert all(
        not isinstance(r, Exception) and r.status_code == 303 for r in results
    ), f"Expected all 303 redirects, got: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"

    # Count successes (no error in redirect URL) vs failures (error in redirect URL)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and "error=" not in r.headers.get("location", ""))
    error_count = sum(1 for r in results if not isinstance(r, Exception) and "error=" in r.headers.get("location", ""))

    # With atomic DELETE ... RETURNING, exactly one should succeed
    assert success_count == 1, f"Expected exactly 1 success, got {success_count}. Redirect URLs: {[r.headers.get('location', '') for r in results if not isinstance(r, Exception)]}"

    # The rest should have errors (tool not found)
    assert error_count == 4, f"Expected 4 errors, got {error_count}. Redirect URLs: {[r.headers.get('location', '') for r in results if not isinstance(r, Exception)]}"

    # Verify tool is actually deleted
    final_list = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
    assert final_list.status_code == 200
    final_tools = final_list.json()["data"]
    assert not any(t["id"] == tool_id for t in final_tools), "Tool should be deleted"


# -------------------------
# Gateway Row Locking Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_gateway_toggle(client: AsyncClient):
    """Test concurrent gateway enable/disable doesn't cause race condition."""
    # Create a gateway with unique URL to avoid conflicts
    unique_id = uuid.uuid4()
    gateway_name = f"Toggle Gateway {unique_id}"
    gateway_data = {"name": gateway_name, "url": f"http://example.com/gateway-{unique_id}", "description": "Toggle test gateway", "visibility": "public", "transport": "SSE"}

    resp = await client.post("/admin/gateways", data=gateway_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200, f"Gateway creation failed with status {resp.status_code}: {resp.text}"

    # Get gateway ID
    list_resp = await client.get("/admin/gateways", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    gateways = list_resp.json()["data"]
    gateway = next((g for g in gateways if g["name"] == gateway_name), None)
    assert gateway is not None
    gateway_id = gateway["id"]

    async def toggle():
        return await client.post(f"/admin/gateways/{gateway_id}/state", data={}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should complete without 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"

    # Verify final state is consistent
    final_resp = await client.get("/admin/gateways", headers=TEST_AUTH_HEADER)
    assert final_resp.status_code == 200


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_team_gateway_creation_same_slug(client: AsyncClient):
    """Test concurrent team gateway creation with same slug prevents duplicates within team."""
    gateway_name = f"Team Gateway {uuid.uuid4()}"

    async def create_team_gateway():
        gateway_data = {"name": gateway_name, "url": "http://example.com/gateway", "description": "Team test gateway", "visibility": "team", "transport": "SSE"}
        return await client.post("/admin/gateways", data=gateway_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name (will generate same slug)
    results = await asyncio.gather(*[create_team_gateway() for _ in range(10)], return_exceptions=True)

    # Count successful creations and conflicts
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code >= 500)

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count >= 8, f"Expected at least 8 conflicts, got {conflict_count}"
    assert error_count == 0, f"Expected no 500 errors, got {error_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_gateway_delete_operations(client: AsyncClient):
    """Test concurrent delete operations with atomic DELETE ... RETURNING.

    With the atomic DELETE ... RETURNING implementation, exactly one delete should
    succeed. All other concurrent deletes will find no row to delete and should
    return an error in the redirect URL.

    Note: The admin endpoint always returns 303 redirects, so we check the
    redirect URL for error messages to distinguish success from failure.
    """
    # Create a gateway with unique URL to avoid conflicts
    unique_id = uuid.uuid4()
    gateway_name = f"Delete Gateway {unique_id}"
    gateway_data = {"name": gateway_name, "url": f"http://example.com/gateway-{unique_id}", "description": "Delete test gateway", "visibility": "public", "transport": "SSE"}

    resp = await client.post("/admin/gateways", data=gateway_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200, f"Gateway creation failed with status {resp.status_code}: {resp.text}"

    # Get gateway ID
    list_resp = await client.get("/admin/gateways", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    gateways = list_resp.json()["data"]
    gateway = next((g for g in gateways if g["name"] == gateway_name), None)
    assert gateway is not None, f"Gateway not found in list. Available gateways: {[g['name'] for g in gateways]}"
    gateway_id = gateway["id"]

    async def delete_gateway():
        return await client.post(f"/admin/gateways/{gateway_id}/delete", data={}, headers=TEST_AUTH_HEADER, follow_redirects=False)

    # Run concurrent deletes
    results = await asyncio.gather(*[delete_gateway() for _ in range(5)], return_exceptions=True)

    # All should return 303 redirects (admin endpoint always redirects)
    assert all(
        not isinstance(r, Exception) and r.status_code == 303 for r in results
    ), f"Expected all 303 redirects, got: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"

    # Count successes (no error in redirect URL) vs failures (error in redirect URL)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and "error=" not in r.headers.get("location", ""))
    error_count = sum(1 for r in results if not isinstance(r, Exception) and "error=" in r.headers.get("location", ""))

    # With atomic DELETE ... RETURNING, at least one should succeed
    # Note: In some cases, gateway deletion might fail due to initialization issues
    # so we check that at least some deletes work correctly
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}. Redirect URLs: {[r.headers.get('location', '') for r in results if not isinstance(r, Exception)]}"

    # If one succeeded, the rest should have errors (gateway not found)
    if success_count == 1:
        assert error_count == 4, f"Expected 4 errors when 1 succeeds, got {error_count}. Redirect URLs: {[r.headers.get('location', '') for r in results if not isinstance(r, Exception)]}"

    # Verify gateway is actually deleted
    final_list = await client.get("/admin/gateways", headers=TEST_AUTH_HEADER)
    assert final_list.status_code == 200
    final_gateways = final_list.json()["data"]
    assert not any(g["id"] == gateway_id for g in final_gateways), "Gateway should be deleted"


# -------------------------
# Skip-Locked Behavior Tests
# -------------------------


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_skip_locked_behavior_tool_updates(client: AsyncClient):
    """Test that skip_locked allows concurrent operations to proceed without blocking."""
    # Create multiple tools
    tool_ids = []
    for i in range(5):
        tool_name = f"skip-lock-tool-{i}-{uuid.uuid4()}"
        tool_data = {
            "name": tool_name,
            "url": f"http://example.com/tool{i}",
            "description": f"Skip lock test tool {i}",
            "integrationType": "REST",
            "requestType": "GET",
            "visibility": "public",
            "headers": "{}",
            "input_schema": "{}",
        }
        resp = await client.post("/admin/tools", data=tool_data, headers=TEST_AUTH_HEADER)
        assert resp.status_code == 200, f"Failed to create tool {i}: {resp.status_code} - {resp.text[:200]}"

        # Get tool ID
        list_resp = await client.get("/admin/tools", headers=TEST_AUTH_HEADER)
        tools = list_resp.json()["data"]
        tool = next((t for t in tools if t["name"] == tool_name), None)
        if tool:
            tool_ids.append(tool["id"])

    async def update_tool(tool_id: str, index: int):
        tool_name = f"updated-tool-{index}-{uuid.uuid4()}"
        update_data = {
            "name": tool_name,
            "customName": tool_name,
            "url": f"http://example.com/updated{index}",
            "requestType": "GET",
            "integrationType": "REST",
            "headers": "{}",
            "input_schema": "{}",
            "description": f"Updated description {index}",
        }
        return await client.post(f"/admin/tools/{tool_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Update all tools concurrently
    results = await asyncio.gather(*[update_tool(tool_id, i) for i, tool_id in enumerate(tool_ids)], return_exceptions=True)

    # Debug: Print all responses
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Tool {i} - Exception: {r}")
        else:
            print(f"Tool {i} - Status: {r.status_code}, Body: {r.text[:200] if hasattr(r, 'text') else 'N/A'}")

    # All should succeed (different tools, skip_locked allows parallel processing)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 303])
    assert success_count == len(tool_ids), f"Expected {len(tool_ids)} successes, got {success_count}. Results: {[(r.status_code if not isinstance(r, Exception) else str(r)) for r in results]}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_mixed_visibility_concurrent_operations(client: AsyncClient):
    """Test concurrent operations with same name - DB constraint enforces uniqueness by team_id+owner_email+name."""
    base_uuid = uuid.uuid4()

    async def create_public_tool():
        # Use unique name per visibility to avoid constraint violation
        form_data = {
            "name": f"mixed-vis-public-{base_uuid}",
            "url": "http://example.com/public",
            "description": "Public tool",
            "integrationType": "REST",
            "requestType": "GET",
            "visibility": "public",
            "headers": "{}",
            "input_schema": "{}",
        }
        return await client.post("/admin/tools", data=form_data, headers=TEST_AUTH_HEADER)

    async def create_team_tool():
        form_data = {
            "name": f"mixed-vis-team-{base_uuid}",
            "url": "http://example.com/team",
            "description": "Team tool",
            "integrationType": "REST",
            "requestType": "GET",
            "visibility": "team",
            "headers": "{}",
            "input_schema": "{}",
        }
        return await client.post("/admin/tools", data=form_data, headers=TEST_AUTH_HEADER)

    async def create_private_tool():
        form_data = {
            "name": f"mixed-vis-private-{base_uuid}",
            "url": "http://example.com/private",
            "description": "Private tool",
            "integrationType": "REST",
            "requestType": "GET",
            "visibility": "private",
            "headers": "{}",
            "input_schema": "{}",
        }
        return await client.post("/admin/tools", data=form_data, headers=TEST_AUTH_HEADER)

    # Create tools with different names and visibility concurrently
    # Each visibility type has 3 concurrent requests with the same name
    results = await asyncio.gather(*[create_public_tool() for _ in range(3)], *[create_team_tool() for _ in range(3)], *[create_private_tool() for _ in range(3)], return_exceptions=True)

    # Should have one success per visibility type (3 total) due to DB constraint on team_id+owner_email+name
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Expect 3 successes (one per unique name) and 6 conflicts (2 per name)
    assert success_count == 3, f"Expected 3 successes (one per unique name), got {success_count}"
    assert conflict_count == 6, f"Expected 6 conflicts, got {conflict_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_high_concurrency_gateway_creation(client: AsyncClient):
    """Test high concurrency with many unique gateway creations."""

    async def create_unique_gateway(index: int):
        gateway_data = {
            "name": f"Concurrent Gateway {index} {uuid.uuid4()}",
            "url": f"http://example{index}.com/gateway",
            "description": f"Concurrent test gateway {index}",
            "visibility": "public",
            "transport": "SSE",
        }
        return await client.post("/admin/gateways", data=gateway_data, headers=TEST_AUTH_HEADER)

    # Create 30 gateways concurrently
    results = await asyncio.gather(*[create_unique_gateway(i) for i in range(30)], return_exceptions=True)

    # All should succeed (different names/slugs)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    failed_count = sum(1 for r in results if isinstance(r, Exception) or (hasattr(r, "status_code") and r.status_code != 200))
    error_500_count = sum(1 for r in results if not isinstance(r, Exception) and hasattr(r, "status_code") and r.status_code >= 500)

    # Allow for occasional failures due to connection pool exhaustion under extreme concurrency
    # At least 28 out of 30 should succeed (93% success rate)
    assert success_count >= 28, f"Expected at least 28 successes, got {success_count} (failed: {failed_count})"

    # No 500 errors - those indicate server bugs, not resource exhaustion
    assert error_500_count == 0, f"Got {error_500_count} server errors (500+), which indicates bugs not resource limits"


# ============================================================================
# PROMPT CONCURRENCY TESTS
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_prompt_creation_same_name(client: AsyncClient):
    """Test concurrent prompt creation with same name prevents duplicates."""
    prompt_name = f"test-prompt-{uuid.uuid4()}"

    async def create_prompt():
        form_data = {"name": prompt_name, "description": "Test prompt", "template": "Test template", "arguments": "[]", "visibility": "public"}
        return await client.post("/admin/prompts", data=form_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name
    results = await asyncio.gather(*[create_prompt() for _ in range(10)], return_exceptions=True)

    # Count successful creations (200) and conflicts (409)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"

    # No 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [200, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_prompt_update_same_name(client: AsyncClient):
    """Test concurrent prompt updates to same name prevents duplicates."""
    # Create two prompts
    prompt1_name = f"prompt-1-{uuid.uuid4()}"
    prompt2_name = f"prompt-2-{uuid.uuid4()}"

    prompt1_data = {"name": prompt1_name, "description": "Prompt 1", "template": "Template 1", "arguments": "[]", "visibility": "public"}
    prompt2_data = {"name": prompt2_name, "description": "Prompt 2", "template": "Template 2", "arguments": "[]", "visibility": "public"}

    resp1 = await client.post("/admin/prompts", data=prompt1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/admin/prompts", data=prompt2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 200
    assert resp2.status_code == 200

    # Get prompt IDs
    list_resp = await client.get("/prompts", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    prompts = list_resp.json()  # Returns list directly, not {"data": [...]}

    prompt1 = next((p for p in prompts if p["name"] == prompt1_name), None)
    prompt2 = next((p for p in prompts if p["name"] == prompt2_name), None)
    assert prompt1 is not None and prompt2 is not None

    prompt1_id = prompt1["id"]
    prompt2_id = prompt2["id"]

    target_name = f"target-prompt-{uuid.uuid4()}"

    async def update_prompt(prompt_id: str):
        update_data = {"name": target_name, "description": "Updated prompt", "template": "Updated template", "arguments": "[]"}
        return await client.post(f"/admin/prompts/{prompt_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both prompts to same name concurrently
    results = await asyncio.gather(*[update_prompt(prompt1_id), update_prompt(prompt2_id)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    _conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)  # noqa: F841

    # At least one should succeed
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_prompt_toggle(client: AsyncClient):
    """Test concurrent enable/disable doesn't cause race condition."""
    prompt_name = f"toggle-prompt-{uuid.uuid4()}"
    prompt_data = {"name": prompt_name, "description": "Toggle test prompt", "template": "Toggle template", "arguments": "[]", "visibility": "public"}

    resp = await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get prompt ID
    list_resp = await client.get("/prompts", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    prompts = list_resp.json()  # Returns list directly, not {"data": [...]}
    prompt = next((p for p in prompts if p["name"] == prompt_name), None)
    assert prompt is not None
    prompt_id = prompt["id"]

    async def toggle():
        return await client.post(f"/admin/prompts/{prompt_id}/state", data={}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should succeed or fail cleanly (no 500 errors)
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_prompt_delete_operations(client: AsyncClient):
    """Test concurrent delete operations handle race conditions correctly."""
    prompt_name = f"delete-prompt-{uuid.uuid4()}"
    prompt_data = {"name": prompt_name, "description": "Delete test prompt", "template": "Delete template", "arguments": "[]", "visibility": "public"}

    resp = await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200

    # Get prompt ID
    list_resp = await client.get("/prompts", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    prompts = list_resp.json()  # Returns list directly, not {"data": [...]}
    prompt = next((p for p in prompts if p["name"] == prompt_name), None)
    assert prompt is not None
    prompt_id = prompt["id"]

    async def delete_prompt():
        return await client.post(f"/admin/prompts/{prompt_id}/delete", data={}, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent deletes
    results = await asyncio.gather(*[delete_prompt() for _ in range(10)], return_exceptions=True)

    # Admin delete always returns 303, but with error in URL if failed
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 303 and "error=" not in r.headers.get("location", ""))
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 303 and "error=" in r.headers.get("location", ""))

    # Exactly one should succeed (admin delete returns 303 redirect without error)
    assert success_count == 1, f"Expected 1 successful delete, got {success_count}"
    assert error_count == 9, f"Expected 9 failed deletes with error, got {error_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_high_concurrency_prompt_creation(client: AsyncClient):
    """Test high concurrency with unique prompts."""

    async def create_unique_prompt(index: int):
        prompt_data = {"name": f"prompt-{uuid.uuid4()}-{index}", "description": f"Prompt {index}", "template": f"Template {index}", "arguments": "[]", "visibility": "public"}
        return await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)

    # Create 50 unique prompts concurrently
    results = await asyncio.gather(*[create_unique_prompt(i) for i in range(50)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    failed_count = sum(1 for r in results if isinstance(r, Exception) or (hasattr(r, "status_code") and r.status_code != 200))

    # Allow for occasional failures due to connection pool exhaustion under extreme concurrency
    # At least 48 out of 50 should succeed (96% success rate)
    assert success_count >= 48, f"Expected at least 48 successful creations, got {success_count} (failed: {failed_count})"


# ============================================================================
# RESOURCE CONCURRENCY TESTS
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_resource_update_same_uri(client: AsyncClient):
    """Test concurrent resource updates to same URI prevents duplicates."""
    # Create two resources
    resource1_uri = f"file:///resource-1-{uuid.uuid4()}.txt"
    resource2_uri = f"file:///resource-2-{uuid.uuid4()}.txt"

    resource1_data = {
        "resource": {"uri": resource1_uri, "name": "Resource 1", "description": "Resource 1", "mimeType": "text/plain", "content": "Resource 1 content"},
        "team_id": None,
        "visibility": "public",
    }
    resource2_data = {
        "resource": {"uri": resource2_uri, "name": "Resource 2", "description": "Resource 2", "mimeType": "text/plain", "content": "Resource 2 content"},
        "team_id": None,
        "visibility": "public",
    }

    resp1 = await client.post("/resources", json=resource1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/resources", json=resource2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 200, f"Failed to create resource 1: {resp1.status_code} - {resp1.text}"
    assert resp2.status_code == 200, f"Failed to create resource 2: {resp2.status_code} - {resp2.text}"

    # Get resource IDs
    list_resp = await client.get("/resources", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    resources = list_resp.json()  # Returns list directly, not wrapped in {"data": [...]}

    resource1 = next((r for r in resources if r["uri"] == resource1_uri), None)
    resource2 = next((r for r in resources if r["uri"] == resource2_uri), None)
    assert resource1 is not None, f"Resource 1 not found in list: {[r['uri'] for r in resources]}"
    assert resource2 is not None, f"Resource 2 not found in list: {[r['uri'] for r in resources]}"

    resource1_id = resource1["id"]
    resource2_id = resource2["id"]

    target_uri = f"file:///target-resource-{uuid.uuid4()}.txt"

    async def update_resource(resource_id: str):
        update_data = {"uri": target_uri, "name": "Updated Resource", "description": "Updated resource", "mimeType": "text/plain"}
        return await client.put(f"/resources/{resource_id}", json=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both resources to same URI concurrently
    results = await asyncio.gather(*[update_resource(resource1_id), update_resource(resource2_id)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Exactly one should succeed, one should conflict
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 1, f"Expected 1 conflict, got {conflict_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_resource_toggle(client: AsyncClient):
    """Test concurrent enable/disable doesn't cause race condition."""
    resource_uri = f"file:///toggle-resource-{uuid.uuid4()}.txt"
    resource_data = {
        "resource": {"uri": resource_uri, "name": "Toggle Resource", "description": "Toggle test resource", "mimeType": "text/plain", "content": "Toggle resource content"},
        "team_id": None,
        "visibility": "public",
    }

    resp = await client.post("/resources", json=resource_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 200, f"Failed to create resource: {resp.status_code} - {resp.text}"

    # Get resource ID
    list_resp = await client.get("/resources", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    resources = list_resp.json()  # Returns list directly
    resource = next((r for r in resources if r["uri"] == resource_uri), None)
    assert resource is not None, f"Resource not found in list: {[r['uri'] for r in resources]}"
    resource_id = resource["id"]

    async def toggle():
        return await client.post(f"/resources/{resource_id}/state", json={}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should succeed or fail cleanly (no 500 errors)
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"

    # Verify final state is consistent
    final_resp = await client.get("/resources", headers=TEST_AUTH_HEADER)
    assert final_resp.status_code == 200
    final_resources = final_resp.json()  # Returns list directly
    final_resource = next((r for r in final_resources if r["uri"] == resource_uri), None)
    assert final_resource is not None, f"Resource not found after toggles: {[r['uri'] for r in final_resources]}"
    # Resource should be in a valid state (either enabled or disabled)
    assert isinstance(final_resource["enabled"], bool)


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_skip_locked_behavior_resource_updates(client: AsyncClient):
    """Test that skip_locked allows concurrent resource operations to proceed without blocking."""
    # Create multiple resources
    resource_ids = []
    for i in range(5):
        resource_uri = f"file:///skip-lock-resource-{i}-{uuid.uuid4()}.txt"
        resource_data = {
            "resource": {"uri": resource_uri, "name": f"Skip lock resource {i}", "description": f"Skip lock test resource {i}", "mimeType": "text/plain", "content": f"Skip lock resource {i} content"},
            "team_id": None,
            "visibility": "public",
        }
        resp = await client.post("/resources", json=resource_data, headers=TEST_AUTH_HEADER)
        assert resp.status_code == 200, f"Failed to create resource {i}: {resp.status_code} - {resp.text}"

        # Get resource ID
        list_resp = await client.get("/resources", headers=TEST_AUTH_HEADER)
        resources = list_resp.json()  # Returns list directly
        resource = next((r for r in resources if r["uri"] == resource_uri), None)
        if resource:
            resource_ids.append(resource["id"])

    async def update_resource(resource_id: str, index: int):
        resource_uri = f"file:///updated-resource-{index}-{uuid.uuid4()}.txt"
        update_data = {"uri": resource_uri, "name": f"Updated resource {index}", "description": f"Updated description {index}", "mimeType": "text/plain"}
        return await client.put(f"/resources/{resource_id}", json=update_data, headers=TEST_AUTH_HEADER)

    # Update all resources concurrently
    results = await asyncio.gather(*[update_resource(resource_id, i) for i, resource_id in enumerate(resource_ids)], return_exceptions=True)

    # All should succeed (different resources, skip_locked allows parallel processing)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    assert success_count == len(resource_ids), f"Expected {len(resource_ids)} successes, got {success_count}"


# ============================================================================
# A2A AGENT CONCURRENCY TESTS
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_a2a_creation_same_name(client: AsyncClient):
    """Test concurrent A2A agent creation with same name prevents duplicates."""
    agent_name = f"test-agent-{uuid.uuid4()}"

    async def create_agent():
        agent_data = {"agent": {"name": agent_name, "description": "Test agent", "endpoint_url": "http://example.com/agent"}, "team_id": None, "visibility": "public"}
        return await client.post("/a2a", json=agent_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name
    results = await asyncio.gather(*[create_agent() for _ in range(10)], return_exceptions=True)

    # Debug: Print all responses
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Result {i}: Exception - {type(r).__name__}: {r}")
        else:
            print(f"Result {i}: Status {r.status_code}")
            if r.status_code not in [201, 409]:
                print(f"  Body: {r.text[:500]}")

    # Count successful creations (201) and conflicts (409)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [201, 409])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    print(f"\nSummary: {success_count} success, {conflict_count} conflicts, {error_count} errors, {exception_count} exceptions")

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"

    # No 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [201, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_a2a_update_same_name(client: AsyncClient):
    """Test concurrent A2A agent updates to same name prevents duplicates."""
    # Create two agents
    agent1_name = f"agent-1-{uuid.uuid4()}"
    agent2_name = f"agent-2-{uuid.uuid4()}"

    agent1_data = {"agent": {"name": agent1_name, "description": "Agent 1", "endpoint_url": "http://example.com/agent1"}, "team_id": None, "visibility": "public"}
    agent2_data = {"agent": {"name": agent2_name, "description": "Agent 2", "endpoint_url": "http://example.com/agent2"}, "team_id": None, "visibility": "public"}

    resp1 = await client.post("/a2a", json=agent1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/a2a", json=agent2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 201, f"Failed to create agent 1: {resp1.status_code} - {resp1.text}"
    assert resp2.status_code == 201, f"Failed to create agent 2: {resp2.status_code} - {resp2.text}"

    # Get agent IDs
    list_resp = await client.get("/a2a", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    agents = list_resp.json()
    if isinstance(agents, dict) and "agents" in agents:
        agents = agents["agents"]

    agent1 = next((a for a in agents if a["name"] == agent1_name), None)
    agent2 = next((a for a in agents if a["name"] == agent2_name), None)
    assert agent1 is not None and agent2 is not None

    agent1_id = agent1["id"]
    agent2_id = agent2["id"]

    target_name = f"target-agent-{uuid.uuid4()}"

    async def update_agent(agent_id: str):
        update_data = {"name": target_name, "description": "Updated agent", "endpoint": "http://example.com/updated"}
        return await client.put(f"/a2a/{agent_id}", json=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both agents to same name concurrently
    results = await asyncio.gather(*[update_agent(agent1_id), update_agent(agent2_id)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # At least one should succeed
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_a2a_toggle(client: AsyncClient):
    """Test concurrent enable/disable doesn't cause race condition."""
    agent_name = f"toggle-agent-{uuid.uuid4()}"
    agent_data = {"agent": {"name": agent_name, "description": "Toggle test agent", "endpoint_url": "http://example.com/agent"}, "team_id": None, "visibility": "public"}

    resp = await client.post("/a2a", json=agent_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 201, f"Failed to create agent: {resp.status_code} - {resp.text}"

    # Get agent ID
    list_resp = await client.get("/a2a", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    agents = list_resp.json()
    if isinstance(agents, dict) and "agents" in agents:
        agents = agents["agents"]
    agent = next((a for a in agents if a["name"] == agent_name), None)
    assert agent is not None
    agent_id = agent["id"]

    async def toggle():
        return await client.post(f"/a2a/{agent_id}/state", json={"activate": True}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should succeed or fail cleanly (no 500 errors)
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_a2a_delete_operations(client: AsyncClient):
    """Test concurrent delete operations handle race conditions correctly."""
    agent_name = f"delete-agent-{uuid.uuid4()}"
    agent_data = {"agent": {"name": agent_name, "description": "Delete test agent", "endpoint_url": "http://example.com/agent"}, "team_id": None, "visibility": "public"}

    resp = await client.post("/a2a", json=agent_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 201, f"Failed to create agent: {resp.status_code} - {resp.text}"

    # Get agent ID
    list_resp = await client.get("/a2a", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    agents = list_resp.json()
    if isinstance(agents, dict) and "agents" in agents:
        agents = agents["agents"]
    agent = next((a for a in agents if a["name"] == agent_name), None)
    assert agent is not None
    agent_id = agent["id"]

    async def delete_agent():
        return await client.delete(f"/a2a/{agent_id}", headers=TEST_AUTH_HEADER)

    # Run 10 concurrent deletes
    results = await asyncio.gather(*[delete_agent() for _ in range(10)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 204])
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 404)

    # Exactly one should succeed
    assert success_count == 1, f"Expected 1 successful delete, got {success_count}"
    assert error_count == 9, f"Expected 9 not found errors, got {error_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_high_concurrency_a2a_creation(client: AsyncClient):
    """Test high concurrency with unique A2A agents."""

    async def create_unique_agent(index: int):
        agent_data = {"name": f"agent-{uuid.uuid4()}-{index}", "description": f"Agent {index}", "endpoint_url": f"http://example.com/agent{index}"}
        return await client.post("/a2a", json={"agent": agent_data, "visibility": "public"}, headers=TEST_AUTH_HEADER)

    # Create 50 unique agents concurrently
    results = await asyncio.gather(*[create_unique_agent(i) for i in range(50)], return_exceptions=True)

    # Debug: Print sample responses
    for i, r in enumerate(results[:3]):  # Print first 3 for debugging
        if isinstance(r, Exception):
            print(f"Result {i}: Exception - {type(r).__name__}: {r}")
        else:
            print(f"Result {i}: Status {r.status_code}")
            if r.status_code != 201:
                print(f"  Body: {r.text[:500]}")

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [201])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    print(f"\nSummary: {success_count} success, {error_count} errors, {exception_count} exceptions")

    # Allow for occasional failures due to connection pool exhaustion under extreme concurrency
    # At least 48 out of 50 should succeed (96% success rate)
    assert success_count >= 48, f"Expected at least 48 successful creations, got {success_count}"


# ============================================================================
# SERVER CONCURRENCY TESTS
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_server_creation_same_name(client: AsyncClient):
    """Test concurrent server creation with same name prevents duplicates."""
    server_name = f"test-server-{uuid.uuid4()}"

    async def create_server():
        server_data = {"name": server_name, "description": "Test server", "transport": "sse", "url": "http://example.com/sse"}
        return await client.post("/servers", json={"server": server_data, "visibility": "public"}, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name
    results = await asyncio.gather(*[create_server() for _ in range(10)], return_exceptions=True)

    # Debug: Print sample responses
    for i, r in enumerate(results[:3]):
        if isinstance(r, Exception):
            print(f"Result {i}: Exception - {type(r).__name__}: {r}")
        else:
            print(f"Result {i}: Status {r.status_code}")
            if r.status_code not in [201, 409]:
                print(f"  Body: {r.text[:500]}")

    # Count successful creations (201) and conflicts (409)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [201, 409])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    print(f"\nSummary: {success_count} success, {conflict_count} conflicts, {error_count} errors, {exception_count} exceptions")

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"

    # No 500 errors
    assert all(isinstance(r, Exception) or r.status_code in [201, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_server_update_same_name(client: AsyncClient):
    """Test concurrent server updates to same name prevents duplicates."""
    # Create two servers
    server1_name = f"server-1-{uuid.uuid4()}"
    server2_name = f"server-2-{uuid.uuid4()}"

    server1_data = {"server": {"name": server1_name, "description": "Server 1", "transport": "sse", "url": "http://example.com/sse1"}, "team_id": None, "visibility": "public"}
    server2_data = {"server": {"name": server2_name, "description": "Server 2", "transport": "sse", "url": "http://example.com/sse2"}, "team_id": None, "visibility": "public"}

    resp1 = await client.post("/servers", json=server1_data, headers=TEST_AUTH_HEADER)
    resp2 = await client.post("/servers", json=server2_data, headers=TEST_AUTH_HEADER)

    assert resp1.status_code == 201, f"Failed to create server 1: {resp1.status_code} - {resp1.text}"
    assert resp2.status_code == 201, f"Failed to create server 2: {resp2.status_code} - {resp2.text}"

    # Get server IDs
    list_resp = await client.get("/servers", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    servers = list_resp.json()  # Returns list directly

    server1 = next((s for s in servers if s["name"] == server1_name), None)
    server2 = next((s for s in servers if s["name"] == server2_name), None)
    assert server1 is not None, f"Server 1 not found in list: {[s['name'] for s in servers]}"
    assert server2 is not None, f"Server 2 not found in list: {[s['name'] for s in servers]}"

    server1_id = server1["id"]
    server2_id = server2["id"]

    target_name = f"target-server-{uuid.uuid4()}"

    async def update_server(server_id: str):
        update_data = {"name": target_name, "description": "Updated server", "transport": "sse", "url": "http://example.com/updated"}
        return await client.put(f"/servers/{server_id}", json=update_data, headers=TEST_AUTH_HEADER)

    # Try to update both servers to same name concurrently
    results = await asyncio.gather(*[update_server(server1_id), update_server(server2_id)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # At least one should succeed
    assert success_count >= 1, f"Expected at least 1 success, got {success_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_server_toggle(client: AsyncClient):
    """Test concurrent enable/disable doesn't cause race condition."""
    server_name = f"toggle-server-{uuid.uuid4()}"
    server_data = {"server": {"name": server_name, "description": "Toggle test server", "transport": "sse", "url": "http://example.com/sse"}, "team_id": None, "visibility": "public"}

    resp = await client.post("/servers", json=server_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 201, f"Failed to create server: {resp.status_code} - {resp.text}"

    # Get server ID
    list_resp = await client.get("/servers", headers=TEST_AUTH_HEADER)
    assert list_resp.status_code == 200
    servers = list_resp.json()  # Returns list directly
    server = next((s for s in servers if s["name"] == server_name), None)
    assert server is not None, f"Server not found in list: {[s['name'] for s in servers]}"
    server_id = server["id"]

    async def toggle():
        return await client.post(f"/servers/{server_id}/state", json={}, headers=TEST_AUTH_HEADER)

    # Run 20 concurrent toggles
    results = await asyncio.gather(*[toggle() for _ in range(20)], return_exceptions=True)

    # All should succeed or fail cleanly (no 500 errors)
    assert all(isinstance(r, Exception) or r.status_code in [200, 303, 404, 409] for r in results), "Some requests returned unexpected status codes"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_server_delete_operations(client: AsyncClient):
    """Test concurrent delete operations handle race conditions correctly."""
    server_name = f"delete-server-{uuid.uuid4()}"
    server_data = {"server": {"name": server_name, "description": "Delete test server", "transport": "sse", "url": "http://example.com/sse"}, "team_id": None, "visibility": "public"}

    resp = await client.post("/servers", json=server_data, headers=TEST_AUTH_HEADER)
    assert resp.status_code == 201, f"Failed to create server: {resp.status_code} - {resp.text}"

    # Get server ID from creation response
    created_server = resp.json()
    server_id = created_server["id"]

    async def delete_server():
        return await client.delete(f"/servers/{server_id}", headers=TEST_AUTH_HEADER)

    # Run 10 concurrent deletes
    results = await asyncio.gather(*[delete_server() for _ in range(10)], return_exceptions=True)

    # Debug: Print sample responses
    for i, r in enumerate(results[:3]):
        if isinstance(r, Exception):
            print(f"Result {i}: Exception - {type(r).__name__}: {r}")
        else:
            print(f"Result {i}: Status {r.status_code}")
            if r.status_code not in [200, 204, 404]:
                print(f"  Body: {r.text[:500]}")

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code in [200, 204])
    error_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 404)
    other_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [200, 204, 404])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    print(f"\nSummary: {success_count} success, {error_count} not found, {other_count} other errors, {exception_count} exceptions")

    # Exactly one should succeed
    assert success_count == 1, f"Expected 1 successful delete, got {success_count}"
    assert error_count == 9, f"Expected 9 not found errors, got {error_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_high_concurrency_server_creation(client: AsyncClient):
    """Test high concurrency with unique servers."""

    async def create_unique_server(index: int):
        server_data = {"name": f"server-{uuid.uuid4()}-{index}", "description": f"Server {index}", "transport": "sse", "url": f"http://example.com/sse{index}"}
        return await client.post("/servers", json={"server": server_data, "visibility": "public"}, headers=TEST_AUTH_HEADER)

    # Create 50 unique servers concurrently
    results = await asyncio.gather(*[create_unique_server(i) for i in range(50)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    failed_count = sum(1 for r in results if isinstance(r, Exception) or (hasattr(r, "status_code") and r.status_code != 201))

    # Allow for occasional failures due to connection pool exhaustion under extreme concurrency
    # At least 48 out of 50 should succeed (96% success rate)
    assert success_count >= 48, f"Expected at least 48 successful creations, got {success_count} (failed: {failed_count})"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_concurrent_team_server_creation_same_name(client: AsyncClient):
    """Test concurrent team server creation with same name prevents duplicates."""
    server_name = f"team-server-{uuid.uuid4()}"

    async def create_team_server():
        server_data = {"server": {"name": server_name, "description": "Team server", "transport": "sse", "url": "http://example.com/sse"}, "team_id": None, "visibility": "team"}
        return await client.post("/servers", json=server_data, headers=TEST_AUTH_HEADER)

    # Run 10 concurrent creations with same name
    results = await asyncio.gather(*[create_team_server() for _ in range(10)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)

    # Exactly one should succeed, rest should be conflicts
    assert success_count == 1, f"Expected 1 success, got {success_count}"
    assert conflict_count == 9, f"Expected 9 conflicts, got {conflict_count}"


# ============================================================================
# SKIP_LOCKED BEHAVIOR TESTS FOR NEW SERVICES
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_skip_locked_behavior_prompt_updates(client: AsyncClient):
    """Test that skip_locked allows concurrent prompt operations to proceed without blocking."""
    # Create multiple prompts
    prompt_ids = []
    for i in range(5):
        prompt_name = f"skip-lock-prompt-{i}-{uuid.uuid4()}"
        prompt_data = {"name": prompt_name, "description": f"Skip lock test prompt {i}", "template": f"Skip lock template {i}", "arguments": "[]", "visibility": "public"}
        resp = await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)
        assert resp.status_code == 200, f"Failed to create prompt {i}: {resp.status_code}"

        # Get prompt ID
        list_resp = await client.get("/prompts", headers=TEST_AUTH_HEADER)
        prompts = list_resp.json()  # Returns list directly, not {"data": [...]}
        prompt = next((p for p in prompts if p["name"] == prompt_name), None)
        if prompt:
            prompt_ids.append(prompt["id"])

    async def update_prompt(prompt_id: str, index: int):
        prompt_name = f"updated-prompt-{index}-{uuid.uuid4()}"
        update_data = {"name": prompt_name, "description": f"Updated description {index}", "template": f"Updated template {index}", "arguments": "[]"}
        return await client.post(f"/admin/prompts/{prompt_id}/edit", data=update_data, headers=TEST_AUTH_HEADER)

    # Update all prompts concurrently
    results = await asyncio.gather(*[update_prompt(prompt_id, i) for i, prompt_id in enumerate(prompt_ids)], return_exceptions=True)

    # All should succeed (different prompts, skip_locked allows parallel processing)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    assert success_count == len(prompt_ids), f"Expected {len(prompt_ids)} successes, got {success_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_skip_locked_behavior_a2a_updates(client: AsyncClient):
    """Test that skip_locked allows concurrent A2A agent operations to proceed without blocking."""
    # Create multiple agents
    agent_ids = []
    for i in range(5):
        agent_name = f"skip-lock-agent-{i}-{uuid.uuid4()}"
        agent_data = {"name": agent_name, "description": f"Skip lock test agent {i}", "endpoint_url": f"http://example.com/agent{i}"}
        resp = await client.post("/a2a", json={"agent": agent_data, "visibility": "public"}, headers=TEST_AUTH_HEADER)
        assert resp.status_code == 201, f"Failed to create agent {i}: {resp.status_code} - {resp.text}"

        # Get agent ID
        list_resp = await client.get("/a2a", headers=TEST_AUTH_HEADER)
        agents = list_resp.json()
        if isinstance(agents, dict) and "agents" in agents:
            agents = agents["agents"]
        agent = next((a for a in agents if a["name"] == agent_name), None)
        if agent:
            agent_ids.append(agent["id"])

    async def update_agent(agent_id: str, index: int):
        agent_name = f"updated-agent-{index}-{uuid.uuid4()}"
        update_data = {"name": agent_name, "description": f"Updated description {index}", "endpoint_url": f"http://example.com/updated{index}"}
        return await client.put(f"/a2a/{agent_id}", json={"agent": update_data}, headers=TEST_AUTH_HEADER)

    # Update all agents concurrently
    results = await asyncio.gather(*[update_agent(agent_id, i) for i, agent_id in enumerate(agent_ids)], return_exceptions=True)

    # All should succeed (different agents, skip_locked allows parallel processing)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    assert success_count == len(agent_ids), f"Expected {len(agent_ids)} successes, got {success_count}"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_skip_locked_behavior_server_updates(client: AsyncClient):
    """Test that skip_locked allows concurrent server operations to proceed without blocking."""
    # Create multiple servers
    server_ids = []
    for i in range(5):
        server_name = f"skip-lock-server-{i}-{uuid.uuid4()}"
        server_data = {"name": server_name, "description": f"Skip lock test server {i}", "transport": "sse", "url": f"http://example.com/sse{i}"}
        resp = await client.post("/servers", json={"server": server_data, "team_id": None, "visibility": "public"}, headers=TEST_AUTH_HEADER)
        assert resp.status_code == 201, f"Failed to create server {i}: {resp.status_code}"

        # Get server ID from creation response
        created_server = resp.json()
        if created_server and "id" in created_server:
            server_ids.append(created_server["id"])

    async def update_server(server_id: str, index: int):
        server_name = f"updated-server-{index}-{uuid.uuid4()}"
        update_data = {"name": server_name, "description": f"Updated description {index}", "transport": "sse", "url": f"http://example.com/updated{index}"}
        return await client.put(f"/servers/{server_id}", json={"server": update_data}, headers=TEST_AUTH_HEADER)

    # Update all servers concurrently
    results = await asyncio.gather(*[update_server(server_id, i) for i, server_id in enumerate(server_ids)], return_exceptions=True)

    # All should succeed (different servers, skip_locked allows parallel processing)
    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    assert success_count == len(server_ids), f"Expected {len(server_ids)} successes, got {success_count}"


# ============================================================================
# MIXED VISIBILITY TESTS FOR NEW SERVICES
# ============================================================================


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_mixed_visibility_concurrent_prompt_operations(client: AsyncClient):
    """Test concurrent prompt operations with different visibility levels."""
    base_uuid = uuid.uuid4()

    async def create_public_prompt():
        prompt_data = {"name": f"mixed-vis-public-prompt-{base_uuid}", "description": "Public prompt", "template": "Public template", "arguments": "[]", "visibility": "public"}
        return await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)

    async def create_team_prompt():
        prompt_data = {"name": f"mixed-vis-team-prompt-{base_uuid}", "description": "Team prompt", "template": "Team template", "arguments": "[]", "visibility": "team"}
        return await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)

    async def create_private_prompt():
        prompt_data = {"name": f"mixed-vis-private-prompt-{base_uuid}", "description": "Private prompt", "template": "Private template", "arguments": "[]", "visibility": "private"}
        return await client.post("/admin/prompts", data=prompt_data, headers=TEST_AUTH_HEADER)

    # Create prompts with different names and visibility concurrently
    results = await asyncio.gather(*[create_public_prompt() for _ in range(3)], *[create_team_prompt() for _ in range(3)], *[create_private_prompt() for _ in range(3)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 200)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    other_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [200, 409])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    # Debug output
    if other_count > 0 or exception_count > 0:
        print(f"\nDebug: {success_count} success, {conflict_count} conflicts, {other_count} other, {exception_count} exceptions")
        for i, r in enumerate(results):
            if isinstance(r, Exception):
                print(f"  Result {i}: Exception - {type(r).__name__}")
            elif r.status_code not in [200, 409]:
                print(f"  Result {i}: Status {r.status_code}")

    # Expect 3 successes (one per unique name) and 6 conflicts
    # Allow for occasional resource exhaustion (timeouts, connection errors)
    assert success_count == 3, f"Expected 3 successes, got {success_count}"
    assert conflict_count >= 4, f"Expected at least 4 conflicts, got {conflict_count} (other: {other_count}, exceptions: {exception_count})"
    assert success_count + conflict_count + other_count + exception_count == 9, "Total should be 9 requests"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_mixed_visibility_concurrent_a2a_operations(client: AsyncClient):
    """Test concurrent A2A agent operations with different visibility levels."""
    base_uuid = uuid.uuid4()

    async def create_public_agent():
        agent_data = {"name": f"mixed-vis-public-agent-{base_uuid}", "description": "Public agent", "endpoint_url": "http://example.com/public"}
        return await client.post("/a2a", json={"agent": agent_data, "visibility": "public"}, headers=TEST_AUTH_HEADER)

    async def create_team_agent():
        agent_data = {"name": f"mixed-vis-team-agent-{base_uuid}", "description": "Team agent", "endpoint_url": "http://example.com/team"}
        return await client.post("/a2a", json={"agent": agent_data, "visibility": "team"}, headers=TEST_AUTH_HEADER)

    async def create_private_agent():
        agent_data = {"name": f"mixed-vis-private-agent-{base_uuid}", "description": "Private agent", "endpoint_url": "http://example.com/private"}
        return await client.post("/a2a", json={"agent": agent_data, "visibility": "private"}, headers=TEST_AUTH_HEADER)

    # Create agents with different names and visibility concurrently
    results = await asyncio.gather(*[create_public_agent() for _ in range(3)], *[create_team_agent() for _ in range(3)], *[create_private_agent() for _ in range(3)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    other_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [201, 409])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    # With different visibility levels and different names per visibility,
    # the uniqueness constraint (team_id, owner_email, slug) allows multiple agents
    # Each visibility level has its own scope, so we expect more successes
    # At minimum, one per unique (name, visibility) combination should succeed
    assert success_count >= 3, f"Expected at least 3 successes, got {success_count}"
    assert success_count + conflict_count + other_count + exception_count == 9, "Total should be 9 requests"


@pytest.mark.asyncio
@SKIP_IF_NOT_POSTGRES
async def test_mixed_visibility_concurrent_server_operations(client: AsyncClient):
    """Test concurrent server operations with different visibility levels."""
    base_uuid = uuid.uuid4()

    async def create_public_server():
        server_data = {"name": f"mixed-vis-public-server-{base_uuid}", "description": "Public server", "transport": "sse", "url": "http://example.com/public"}
        return await client.post("/servers", json={"server": server_data, "team_id": None, "visibility": "public"}, headers=TEST_AUTH_HEADER)

    async def create_team_server():
        server_data = {"name": f"mixed-vis-team-server-{base_uuid}", "description": "Team server", "transport": "sse", "url": "http://example.com/team"}
        return await client.post("/servers", json={"server": server_data, "team_id": None, "visibility": "team"}, headers=TEST_AUTH_HEADER)

    async def create_private_server():
        server_data = {"name": f"mixed-vis-private-server-{base_uuid}", "description": "Private server", "transport": "sse", "url": "http://example.com/private"}
        return await client.post("/servers", json={"server": server_data, "team_id": None, "visibility": "private"}, headers=TEST_AUTH_HEADER)

    # Create servers with different names and visibility concurrently
    results = await asyncio.gather(*[create_public_server() for _ in range(3)], *[create_team_server() for _ in range(3)], *[create_private_server() for _ in range(3)], return_exceptions=True)

    success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
    conflict_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 409)
    other_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code not in [201, 409])
    exception_count = sum(1 for r in results if isinstance(r, Exception))

    # With different visibility levels and different names per visibility,
    # the uniqueness constraint (team_id, owner_email, name) allows multiple servers
    # Each visibility level has its own scope, so we expect more successes
    # At minimum, one per unique (name, visibility) combination should succeed
    assert success_count >= 3, f"Expected at least 3 successes, got {success_count}"
    assert success_count + conflict_count + other_count + exception_count == 9, "Total should be 9 requests"
