# -*- coding: utf-8 -*-
"""Location: ./tests/unit/mcpgateway/services/test_prompt_service.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Unit-tests for PromptService.
All tests run entirely with `MagicMock` / `AsyncMock`; no live DB or Jinja
environment is required.  Where `PromptService` returns Pydantic models we
monkey-patch the `model_validate` method so that it simply echoes the raw
dict we pass in - that keeps validation out of scope for these pure-unit
tests.
"""

# Future
from __future__ import annotations

# Standard
from datetime import datetime, timezone
from types import SimpleNamespace
from typing import Any, List, Optional, TypeVar
from unittest.mock import AsyncMock, MagicMock, Mock, patch

# Third-Party
import pytest
from sqlalchemy.exc import IntegrityError

# First-Party
from mcpgateway.common.models import Message, PromptResult, Role, TextContent
from mcpgateway.db import Prompt as DbPrompt
from mcpgateway.db import PromptMetric
from mcpgateway.schemas import PromptArgument, PromptCreate, PromptMetrics, PromptRead, PromptUpdate
from mcpgateway.services.prompt_service import PromptError, PromptNameConflictError, PromptNotFoundError, PromptService, PromptValidationError

# Local
from tests.helpers.admin_mocks import install_admin_user

# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def mock_logging_services():
    """Mock audit_trail and structured_logger to prevent database writes during tests."""
    with patch("mcpgateway.services.prompt_service.audit_trail") as mock_audit, patch("mcpgateway.services.prompt_service.structured_logger") as mock_logger:
        mock_audit.log_action = MagicMock(return_value=None)
        mock_logger.log = MagicMock(return_value=None)
        yield {"audit_trail": mock_audit, "structured_logger": mock_logger}


@pytest.fixture
def mock_prompt():
    """Create a mock prompt model."""
    prompt = MagicMock()

    prompt.id = "1"
    prompt.name = "test"
    prompt.description = "Test prompt"
    prompt.template = "Hello!"
    prompt.argument_schema = {}
    prompt.version = 1
    prompt.visibility = "public"
    prompt.team_id = None
    prompt.owner_email = None

    return prompt


_R = TypeVar("_R")


def _make_execute_result(*, scalar: Any = _R | None, scalars_list: list[_R] | None = None) -> MagicMock:
    """
    Return a MagicMock that mimics the SQLAlchemy Result object:

      - .scalar_one_or_none() → scalar
      - .scalar()            → scalar
      - .scalars().all()     → scalars_list
    """
    result = MagicMock()
    result.scalar_one_or_none.return_value = scalar
    result.scalar.return_value = scalar
    scalars_proxy = MagicMock()
    scalars_proxy.all.return_value = scalars_list or []
    result.scalars.return_value = scalars_proxy
    return result


def _build_db_prompt(
    *,
    pid: int = 1,
    name: str = "hello",
    desc: str = "greeting",
    template: str = "Hello, {{ name }}!",
    is_active: bool = True,
    metrics: Optional[List[PromptMetric]] = None,
) -> MagicMock:
    """Return a MagicMock that looks like a DbPrompt instance."""
    p = MagicMock(spec=DbPrompt)
    p.id = pid
    p.name = name
    p.original_name = name
    p.custom_name = name
    p.custom_name_slug = name
    p.display_name = name
    p.description = desc
    p.template = template
    p.argument_schema = {"properties": {"name": {"type": "string"}}, "required": ["name"]}
    p.created_at = p.updated_at = datetime(2025, 1, 1, tzinfo=timezone.utc)
    p.is_active = is_active
    # New model uses `enabled` — keep both attributes for backward compatibility in tests
    p.enabled = is_active
    p.visibility = "public"
    p.team_id = None
    p.owner_email = "owner@example.com"
    p.gateway_id = None
    p.gateway = None
    p.metrics = metrics or []

    # Mock metrics_summary property to match the new implementation
    # Calculate summary from provided metrics
    if metrics:
        total = len(metrics)
        successful = sum(1 for m in metrics if m.is_success)
        failed = total - successful
        failure_rate = failed / total if total > 0 else 0.0
        min_rt = min((m.response_time for m in metrics), default=None)
        max_rt = max((m.response_time for m in metrics), default=None)
        avg_rt = sum(m.response_time for m in metrics) / total if total > 0 else None
        last_time = max((m.timestamp for m in metrics), default=None)

        p.metrics_summary = {
            "total_executions": total,
            "successful_executions": successful,
            "failed_executions": failed,
            "failure_rate": failure_rate,
            "min_response_time": min_rt,
            "max_response_time": max_rt,
            "avg_response_time": avg_rt,
            "last_execution_time": last_time,
        }
    else:
        p.metrics_summary = {
            "total_executions": 0,
            "successful_executions": 0,
            "failed_executions": 0,
            "failure_rate": 0.0,
            "min_response_time": None,
            "max_response_time": None,
            "avg_response_time": None,
            "last_execution_time": None,
        }

    # validate_arguments: accept anything
    p.validate_arguments = Mock()
    return p


# ---------------------------------------------------------------------------
# auto-use fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def _patch_promptread(monkeypatch):
    """
    Bypass Pydantic validation: make PromptRead.model_validate a pass-through.
    """
    monkeypatch.setattr(PromptRead, "model_validate", staticmethod(lambda d: d))


@pytest.fixture(autouse=True)
def reset_jinja_singleton():
    """Reset the module-level Jinja environment singleton before each test.

    This is needed because PromptService now uses a shared singleton for
    the Jinja environment (for caching), so tests that modify the environment
    can affect subsequent tests.
    """
    # First-Party
    import mcpgateway.services.prompt_service as ps

    ps._JINJA_ENV = None
    ps._compile_jinja_template.cache_clear()
    yield
    ps._JINJA_ENV = None
    ps._compile_jinja_template.cache_clear()


# ---------------------------------------------------------------------------
# main service fixture
# ---------------------------------------------------------------------------


@pytest.fixture
def prompt_service():
    svc = PromptService()
    return svc


# ---------------------------------------------------------------------------
# TESTS
# ---------------------------------------------------------------------------


class TestPromptServiceInit:
    @pytest.mark.asyncio
    async def test_plugins_enabled_env_flag_false_disables_plugin_manager(self, monkeypatch):
        """Cover env-override parsing in PromptService._get_plugin_manager (PLUGINS_ENABLED)."""
        from mcpgateway.plugins import enable_plugins, reset_plugin_manager_factory


        monkeypatch.setenv("PLUGINS_ENABLED", "false")
        enable_plugins(False)
        reset_plugin_manager_factory()

        svc = PromptService()
        result = await svc._get_plugin_manager(None)
        assert not result


class TestPromptService:
    # ──────────────────────────────────────────────────────────────────
    #   register_prompt
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_register_prompt_success(self, prompt_service, test_db):
        """Happy-path prompt registration."""
        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()

        prompt_service._notify_prompt_added = AsyncMock()

        pc = PromptCreate(
            name="hello",
            description="greet a user",
            template="Hello {{ name }}!",
            arguments=[],
        )

        res = await prompt_service.register_prompt(test_db, pc)

        test_db.add.assert_called_once()
        test_db.commit.assert_called_once()
        test_db.refresh.assert_called_once()
        prompt_service._notify_prompt_added.assert_called_once()
        assert res["name"] == "hello"
        assert res["template"] == "Hello {{ name }}!"

    @pytest.mark.asyncio
    async def test_register_prompt_argument_description_included_in_schema(self, prompt_service, test_db):
        """Cover argument.description -> schema['description'] branch."""
        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()
        prompt_service._notify_prompt_added = AsyncMock()

        with (
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
            patch("mcpgateway.cache.metrics_cache.metrics_cache") as mock_metrics_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache.invalidate_prompts = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            mock_metrics_cache.invalidate_prefix = Mock()
            mock_metrics_cache.invalidate = Mock()

            pc = PromptCreate(
                name="hello",
                description="greet a user",
                template="Hello {{ name }}!",
                arguments=[PromptArgument(name="name", description="who")],
            )

            res = await prompt_service.register_prompt(test_db, pc)

        assert res["arguments"][0]["description"] == "who"
        assert res["arguments"][0]["required"] is True

    @pytest.mark.asyncio
    async def test_register_prompt_gateway_relationship_is_set(self, prompt_service, test_db):
        """Cover gateway relationship assignment when gateway_id resolves to a gateway."""
        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.name = "Gateway One"

        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=gateway),  # gateway lookup
                _make_execute_result(scalar=None),  # conflict check
            ]
        )
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()
        prompt_service._notify_prompt_added = AsyncMock()

        with (
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
            patch("mcpgateway.cache.metrics_cache.metrics_cache") as mock_metrics_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache.invalidate_prompts = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            mock_metrics_cache.invalidate_prefix = Mock()
            mock_metrics_cache.invalidate = Mock()

            pc = PromptCreate(
                name="hello",
                description="greet a user",
                template="Hello!",
                arguments=[],
                gateway_id="gw-1",
            )

            _ = await prompt_service.register_prompt(test_db, pc)

        added = test_db.add.call_args.args[0]
        assert added.gateway is gateway
        assert getattr(added, "gateway_name_cache") == "Gateway One"

    @pytest.mark.asyncio
    async def test_register_prompt_conflict(self, prompt_service, test_db):
        """Existing prompt with same name → PromptNameConflictError."""
        existing = _build_db_prompt()
        test_db.execute = Mock(return_value=_make_execute_result(scalar=existing))

        pc = PromptCreate(name="hello", description="", template="X", arguments=[])

        try:
            await prompt_service.register_prompt(test_db, pc)
            assert False, "Expected PromptError for duplicate prompt name"
        except PromptError as exc:
            msg = str(exc)
            # Simulate a response-like error dict for message checking
            # (since this is a unit test, we only have the exception message)
            if "detail" in msg:
                assert "already exists" in msg
            elif "message" in msg:
                assert "already exists" in msg
            else:
                # Accept any error format as long as status is correct
                assert "409" in msg or "already exists" in msg or "Failed to register prompt" in msg

    @pytest.mark.asyncio
    async def test_register_prompt_team_visibility_conflict_hits_team_branch(self, prompt_service, test_db):
        """Existing team prompt with same name+team_id triggers the team conflict branch."""
        existing = MagicMock()
        existing.enabled = True
        existing.id = 123
        existing.visibility = "team"
        test_db.execute = Mock(return_value=_make_execute_result(scalar=existing))
        test_db.rollback = Mock()

        pc = PromptCreate(name="hello", description="", template="X", arguments=[])

        with pytest.raises(PromptNameConflictError):
            await prompt_service.register_prompt(test_db, pc, visibility="team", team_id="team-1")

    @pytest.mark.asyncio
    async def test_register_prompt_slug_conflict(self, prompt_service, test_db):
        """Slug collisions should be detected as name conflicts."""
        existing = _build_db_prompt(name="hello-world")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=existing))

        pc = PromptCreate(name="Hello World", description="", template="X", arguments=[])

        with pytest.raises(PromptError):
            await prompt_service.register_prompt(test_db, pc)

    @pytest.mark.asyncio
    async def test_register_prompt_template_validation_error(self, prompt_service, test_db):
        """Test that template validation errors are raised as TemplateValidationError."""
        # First-Party
        from mcpgateway.services.content_security import TemplateValidationError

        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()
        prompt_service._notify_prompt_added = AsyncMock()

        # Use a template with nonexistent filter (passes Pydantic, fails service validation)
        pc = PromptCreate(name="fail", description="", template="Hello {{ name | nonexistent_filter }}", arguments=[])

        with pytest.raises(TemplateValidationError) as exc_info:
            await prompt_service.register_prompt(test_db, pc)
        # Generic error message to avoid leaking template details (CWE-209 fix)
        assert "invalid jinja2 syntax" in str(exc_info.value).lower()

    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        "err_msg",
        [
            "UNIQUE constraint failed: prompt.name",  # duplicate name
            "CHECK constraint failed: prompt",  # check constraint
            "NOT NULL constraint failed: prompt.name",  # not null
        ],
    )
    async def test_register_prompt_integrity_error(self, prompt_service, test_db, err_msg):
        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()
        prompt_service._notify_prompt_added = AsyncMock()
        test_db.commit.side_effect = IntegrityError(err_msg, None, BaseException(None))
        pc = PromptCreate(name="fail", description="", template="ok", arguments=[])
        with pytest.raises(IntegrityError) as exc_info:
            await prompt_service.register_prompt(test_db, pc)
        msg = str(exc_info.value).lower()
        assert err_msg.lower() in msg

    # ──────────────────────────────────────────────────────────────────
    #   get_prompt
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_register_prompt_content_size_error(self, prompt_service, test_db):
        """Test that ContentSizeError is caught and re-raised during prompt registration."""
        # First-Party
        from mcpgateway.services.content_security import ContentSizeError

        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises ContentSizeError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_size.side_effect = ContentSizeError(content_type="Prompt template", actual_size=15000, max_size=10240)

        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service):
            # Use 15KB template - passes Pydantic (65KB limit) but fails ContentSizeError (10KB limit)
            prompt = PromptCreate(
                name="large-prompt",
                description="A prompt with large template",
                template="x" * 15000,  # 15KB template
            )

            with pytest.raises(ContentSizeError) as exc_info:
                await prompt_service.register_prompt(
                    test_db,
                    prompt,
                    created_by="user@example.com",
                    owner_email="user@example.com",
                )

            # Verify the error details
            assert exc_info.value.actual_size == 15000
            assert exc_info.value.max_size == 10240
            assert exc_info.value.content_type == "Prompt template"

            # Verify rollback was called
            test_db.rollback.assert_called_once()

    @pytest.mark.asyncio
    async def test_get_prompt_with_metadata(self, prompt_service, test_db):
        """Test get_prompt accepts metadata."""
        db_prompt = _build_db_prompt(template="Hello!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        meta_data = {"trace_id": "123"}

        # Just verify it doesn't crash and returns result
        result = await prompt_service.get_prompt(test_db, "1", {}, _meta_data=meta_data)
        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_rendered(self, prompt_service, test_db):
        """Prompt is fetched and rendered into Message objects."""
        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        pr: PromptResult = await prompt_service.get_prompt(test_db, "1", {"name": "Alice"})

        assert isinstance(pr, PromptResult)
        assert len(pr.messages) == 1
        msg: Message = pr.messages[0]
        assert msg.role == Role.USER
        assert isinstance(msg.content, TextContent)
        assert msg.content.text == "Hello, Alice!"

    @pytest.mark.asyncio
    async def test_get_prompt_gateway_backed_blank_template_fetches_remote_result(self, prompt_service, test_db):
        """Gateway-backed prompts without local templates should execute upstream."""
        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.url = "http://gateway.example.com/mcp"
        gateway.transport = "streamable_http"

        db_prompt = _build_db_prompt(
            name="fast-time-convert-time-detailed",
            template="",
            desc="Convert time with detailed context",
        )
        db_prompt.original_name = "convert_time_detailed"
        db_prompt.gateway_id = gateway.id
        db_prompt.gateway = gateway

        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))
        test_db.commit = Mock()
        # Mock _apply_access_control to return the query unchanged (simulates access granted)
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        remote_result = PromptResult(
            messages=[
                Message(
                    role=Role.USER,
                    content=TextContent(
                        type="text",
                        text="Rendered prompt for America/New_York and Europe/Dublin",
                    ),
                )
            ],
            description="Convert time with detailed context",
        )
        prompt_service._fetch_gateway_prompt_result = AsyncMock(return_value=remote_result)

        result = await prompt_service.get_prompt(
            test_db,
            db_prompt.name,
            {"from_timezone": "UTC", "to_timezones": "America/New_York,Europe/Dublin"},
            user="user@test.com",
        )

        prompt_service._fetch_gateway_prompt_result.assert_awaited_once_with(
            db_prompt,
            {"from_timezone": "UTC", "to_timezones": "America/New_York,Europe/Dublin"},
            meta_data=None,
        )
        test_db.commit.assert_called_once()
        assert result.description == "Convert time with detailed context"
        assert result.messages[0].content.text == "Rendered prompt for America/New_York and Europe/Dublin"

    @pytest.mark.asyncio
    async def test_get_prompt_by_name(self, prompt_service, test_db):
        """Prompt lookup prioritizes name first (MCP spec), then falls back to ID."""
        db_prompt = _build_db_prompt(template="Hello!")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        result = await prompt_service.get_prompt(test_db, "gateway__greeting", {})
        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_name_lookup_priority(self, prompt_service, test_db):
        """Issue #1704: Verify name-based lookup is prioritized over ID lookup per MCP spec.

        This test ensures that when a prompt is looked up, the service tries to find it
        by name first (MCP specification requirement), and only falls back to ID lookup
        if the name lookup fails (for backward compatibility).
        """
        db_prompt = _build_db_prompt(name="compare_timezones", template="Hello!")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        result = await prompt_service.get_prompt(test_db, "compare_timezones", {})
        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_id_fallback_backward_compat(self, prompt_service, test_db):
        """Verify ID-based lookup still works as fallback for backward compatibility."""
        db_prompt = _build_db_prompt(pid=123, name="some_prompt", template="Hello!")
        db_prompt.enabled = True  # Ensure prompt is active
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)
        # Single execute call - _find_prompt_by_name_or_id uses OR query (name OR id)
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        result = await prompt_service.get_prompt(test_db, "123", {})
        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_multiple_results_falls_back_to_name(self, prompt_service, test_db):
        """When OR(name, id) matches multiple rows, fall back to name-only lookup."""
        # Third-Party
        from sqlalchemy.exc import MultipleResultsFound

        db_prompt = _build_db_prompt(name="collision_id", template="By name!")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        first_result = MagicMock()
        first_result.scalar_one_or_none.side_effect = MultipleResultsFound()

        second_result = _make_execute_result(scalar=db_prompt)
        test_db.execute = Mock(side_effect=[first_result, second_result])

        result = await prompt_service.get_prompt(test_db, "collision_id", {})
        assert result.messages[0].content.text == "By name!"

    @pytest.mark.asyncio
    async def test_get_prompt_ambiguous_name_raises_prompt_error(self, prompt_service, test_db):
        """When multiple accessible prompts share the same name, raise explicit ambiguity error."""
        # Third-Party
        from sqlalchemy.exc import MultipleResultsFound

        # First-Party
        from mcpgateway.services.prompt_service import PromptError

        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        # Both OR query and name-only fallback hit multiple rows
        or_result = MagicMock()
        or_result.scalar_one_or_none.side_effect = MultipleResultsFound()
        name_result = MagicMock()
        name_result.scalar_one_or_none.side_effect = MultipleResultsFound()
        test_db.execute = Mock(side_effect=[or_result, name_result])

        with pytest.raises(PromptError, match="ambiguous"):
            await prompt_service.get_prompt(test_db, "code_review", {})

    @pytest.mark.asyncio
    async def test_get_prompt_multi_team_same_name(self, prompt_service, test_db):
        """Issue #1704: Verify team scoping prevents cross-team prompt access with duplicate names.

        When multiple teams have prompts with the same name, the service should:
        1. Apply team scoping BEFORE lookup to find the correct team's prompt
        2. Not return another team's prompt even if it has the same name
        3. Respect the user's token_teams for access control
        """
        # Team A's prompt with name "code_review"
        team_a_prompt = _build_db_prompt(pid="prompt-team-a", name="code_review", template="Team A template")
        team_a_prompt.team_id = "team-a"
        team_a_prompt.owner_email = "owner-a@test.com"
        team_a_prompt.visibility = "team"

        # Team B's prompt with same name "code_review"
        team_b_prompt = _build_db_prompt(pid="prompt-team-b", name="code_review", template="Team B template")
        team_b_prompt.team_id = "team-b"
        team_b_prompt.owner_email = "owner-b@test.com"
        team_b_prompt.visibility = "team"

        # Mock _apply_access_control to return scoped query for team B
        async def mock_apply_access_control(query, db, user, token_teams, team_id):
            # Simulate filtering to only team B's prompts
            # First-Party
            from mcpgateway.db import Prompt as DbPrompt

            return query.where(DbPrompt.team_id == "team-b")

        with patch.object(prompt_service, "_apply_access_control", side_effect=mock_apply_access_control):
            # User from Team B requests "code_review"
            test_db.execute = Mock(return_value=_make_execute_result(scalar=team_b_prompt))

            result = await prompt_service.get_prompt(test_db, "code_review", {}, user="member-b@test.com", token_teams=["team-b"])

            # Should get Team B's template, not Team A's
            assert result.messages[0].content.text == "Team B template"

    @pytest.mark.asyncio
    async def test_get_prompt_not_found(self, prompt_service, test_db):
        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))

        with pytest.raises(PromptNotFoundError):
            await prompt_service.get_prompt(test_db, "999")

    @pytest.mark.asyncio
    async def test_get_prompt_inactive(self, prompt_service, test_db):
        inactive = _build_db_prompt(is_active=False)
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=None),  # active query (OR name/id) - not found
                _make_execute_result(scalar=inactive),  # inactive query (OR name/id) - found
            ]
        )
        with pytest.raises(PromptNotFoundError) as exc_info:
            await prompt_service.get_prompt(test_db, "1")
        assert "inactive" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_get_prompt_render_error(self, prompt_service, test_db):
        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))
        db_prompt.validate_arguments.side_effect = Exception("bad args")
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.get_prompt(test_db, "1", {"name": "Alice"})
        assert "Failed to process prompt" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_get_prompt_observability_start_span_exception_is_swallowed(self, prompt_service, test_db):
        """Covers start_span() exception path (should not break prompt rendering)."""
        # Standard
        from contextlib import contextmanager

        db_prompt = _build_db_prompt(template="Hello!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        obs = MagicMock()
        obs.start_span.side_effect = Exception("boom")

        @contextmanager
        def _no_span(*_a, **_kw):
            yield None

        with (
            patch("mcpgateway.services.prompt_service.current_trace_id") as mock_trace,
            patch("mcpgateway.services.prompt_service.ObservabilityService", return_value=obs),
            patch("mcpgateway.services.prompt_service.create_span", _no_span),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf,
        ):
            mock_trace.get.return_value = "trace-1"
            mock_get_buf.record_prompt_metric = Mock()
            result = await prompt_service.get_prompt(test_db, "1", {})

        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_observability_end_span_success_and_metrics_error(self, prompt_service, test_db):
        """Covers observability span start+end plus metrics buffer exception handling."""
        # Standard
        from contextlib import contextmanager

        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        span = MagicMock()

        @contextmanager
        def _span_cm(*_a, **_kw):
            yield span

        obs = MagicMock()
        obs.start_span.return_value = "span-1"
        obs.end_span = Mock()

        metrics_buffer = MagicMock()
        metrics_buffer.record_prompt_metric.side_effect = Exception("metrics boom")

        with (
            patch("mcpgateway.services.prompt_service.current_trace_id") as mock_trace,
            patch("mcpgateway.services.prompt_service.ObservabilityService", return_value=obs),
            patch("mcpgateway.services.prompt_service.create_span", _span_cm),
            patch("mcpgateway.services.prompt_service.metrics_buffer", metrics_buffer),
        ):
            mock_trace.get.return_value = "trace-1"
            result = await prompt_service.get_prompt(test_db, "1", {"name": "Alice"})

        assert result.messages[0].content.text == "Hello, Alice!"
        obs.end_span.assert_called_once()
        span.set_attribute.assert_any_call("success", True)

    @pytest.mark.asyncio
    async def test_get_prompt_sets_langfuse_prompt_linkage_attributes(self, prompt_service, test_db):
        """Successful prompt rendering should link the Langfuse observation to the prompt version."""
        # Standard
        from contextlib import contextmanager

        db_prompt = _build_db_prompt(name="greeting_prompt", template="Hello!")
        db_prompt.version = 7
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        span = MagicMock()

        @contextmanager
        def _span_cm(*_a, **_kw):
            yield span

        with (
            patch("mcpgateway.services.prompt_service.create_span", _span_cm),
            patch("mcpgateway.services.prompt_service.is_output_capture_enabled", return_value=True),
            patch("mcpgateway.services.prompt_service.set_span_attribute", side_effect=lambda target, key, value: target.set_attribute(key, value)),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_metrics_buffer,
        ):
            mock_metrics_buffer.record_prompt_metric = Mock()
            result = await prompt_service.get_prompt(test_db, "1", {})

        assert result.messages[0].content.text == "Hello!"
        span.set_attribute.assert_any_call("langfuse.observation.prompt.name", "greeting_prompt")
        span.set_attribute.assert_any_call("langfuse.observation.prompt.version", 7)
        assert any(call.args[0] == "langfuse.observation.output" for call in span.set_attribute.call_args_list)

    @pytest.mark.asyncio
    async def test_get_prompt_observability_end_span_exception_is_caught(self, prompt_service, test_db):
        # Standard
        from contextlib import contextmanager

        db_prompt = _build_db_prompt(template="Hello!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        @contextmanager
        def _no_span(*_a, **_kw):
            yield None

        obs = MagicMock()
        obs.start_span.return_value = "span-1"
        obs.end_span.side_effect = Exception("end boom")

        with (
            patch("mcpgateway.services.prompt_service.current_trace_id") as mock_trace,
            patch("mcpgateway.services.prompt_service.ObservabilityService", return_value=obs),
            patch("mcpgateway.services.prompt_service.create_span", _no_span),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf,
        ):
            mock_trace.get.return_value = "trace-1"
            mock_get_buf.record_prompt_metric = Mock()
            result = await prompt_service.get_prompt(test_db, "1", {})

        assert result.messages[0].content.text == "Hello!"

    @pytest.mark.asyncio
    async def test_get_prompt_access_denied_raises_generic_not_found(self, prompt_service, test_db):
        """Access control is now applied via scoped query, so denied prompts are not found."""

        # Mock _apply_access_control to return a query that filters out all prompts
        async def mock_apply_access_control(query, db, user, token_teams, team_id):
            # First-Party
            from mcpgateway.db import Prompt as DbPrompt

            # Return query that will never match (simulates access denial)
            return query.where(DbPrompt.id == "nonexistent")

        with patch.object(prompt_service, "_apply_access_control", side_effect=mock_apply_access_control):
            test_db.execute = Mock(return_value=_make_execute_result(scalar=None))

            with patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf:
                mock_get_buf.record_prompt_metric = Mock()
                with pytest.raises(PromptNotFoundError, match="Prompt not found"):
                    await prompt_service.get_prompt(test_db, "1", {})

    @pytest.mark.asyncio
    async def test_get_prompt_server_scoping_not_attached_raises_not_found(self, prompt_service, test_db):
        db_prompt = _build_db_prompt(template="Hello!")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        server_match_result = MagicMock()
        server_match_result.first.return_value = None
        test_db.execute = Mock(side_effect=[_make_execute_result(scalar=db_prompt), server_match_result])

        with patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf:
            mock_get_buf.record_prompt_metric = Mock()
            with pytest.raises(PromptNotFoundError, match="Prompt not found"):
                await prompt_service.get_prompt(test_db, "1", {}, server_id="server-1")

    @pytest.mark.asyncio
    async def test_get_prompt_server_id_disambiguates_duplicate_names(self, prompt_service, test_db):
        """When server_id is provided, it scopes the lookup so duplicate names across servers resolve correctly."""
        db_prompt = _build_db_prompt(name="shared_prompt", template="Server 1 version")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        # First call: server-joined lookup returns the prompt; second call: post-lookup server check
        server_match_result = MagicMock()
        server_match_result.first.return_value = ("ok",)
        test_db.execute = Mock(side_effect=[_make_execute_result(scalar=db_prompt), server_match_result])

        result = await prompt_service.get_prompt(test_db, "shared_prompt", {}, server_id="server-1")
        assert result.messages[0].content.text == "Server 1 version"

    @pytest.mark.asyncio
    async def test_get_prompt_server_id_inactive_scoped(self, prompt_service, test_db):
        """Inactive prompt lookup also applies server_id scoping."""
        inactive = _build_db_prompt(is_active=False, name="disabled_prompt")
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=None),  # active server-scoped query - not found
                _make_execute_result(scalar=inactive),  # inactive server-scoped query - found
            ]
        )
        with pytest.raises(PromptNotFoundError, match="inactive"):
            await prompt_service.get_prompt(test_db, "disabled_prompt", {}, server_id="server-1")

    @pytest.mark.asyncio
    async def test_get_prompt_render_error_sets_span_attributes(self, prompt_service, test_db):
        """Covers span attribute setting on render errors."""
        # Standard
        from contextlib import contextmanager

        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))
        db_prompt.validate_arguments.side_effect = Exception("bad args")

        span = MagicMock()

        @contextmanager
        def _span_cm(*_a, **_kw):
            yield span

        with (
            patch("mcpgateway.services.prompt_service.create_span", _span_cm),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf,
        ):
            mock_get_buf.record_prompt_metric = Mock()
            with pytest.raises(PromptError, match="Failed to process prompt"):
                await prompt_service.get_prompt(test_db, "1", {"name": "Alice"})

        span.set_attribute.assert_any_call("error", True)

    @pytest.mark.asyncio
    async def test_get_prompt_plugin_pre_and_post_hooks_modify_payload_and_result(self, prompt_service, test_db):
        """Covers plugin hook paths including GlobalContext update when provided."""
        # Standard
        from contextlib import contextmanager

        # First-Party
        from cpex.framework import GlobalContext, PromptHookType, PromptPosthookPayload, PromptPrehookPayload

        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")

        server_match_result = MagicMock()
        server_match_result.first.return_value = ("ok",)

        test_db.execute = Mock(side_effect=[_make_execute_result(scalar=db_prompt), server_match_result])
        prompt_service._apply_access_control = AsyncMock(side_effect=lambda q, *args, **kwargs: q)

        plugin_mgr = MagicMock()
        plugin_mgr.has_hooks_for.side_effect = lambda hook: hook in {PromptHookType.PROMPT_PRE_FETCH, PromptHookType.PROMPT_POST_FETCH}

        pre_payload = PromptPrehookPayload(prompt_id="1", args={"name": "Alice"})
        pre_result = SimpleNamespace(modified_payload=pre_payload)

        modified = PromptResult(messages=[Message(role=Role.ASSISTANT, content=TextContent(type="text", text="post"))], description="post")
        post_payload = PromptPosthookPayload(prompt_id="1", result=modified)
        post_result = SimpleNamespace(modified_payload=post_payload)

        plugin_mgr.invoke_hook = AsyncMock(side_effect=[(pre_result, {"ctx": 1}), (post_result, {"ctx": 1})])

        @contextmanager
        def _no_span(*_a, **_kw):
            yield None

        global_ctx = GlobalContext(request_id="req-1")

        with (
            patch("mcpgateway.services.prompt_service.create_span", _no_span),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf,
            patch.object(prompt_service, "_get_plugin_manager", AsyncMock(return_value=plugin_mgr)),
        ):
            mock_get_buf.record_prompt_metric = Mock()
            result = await prompt_service.get_prompt(
                test_db,
                "1",
                {"name": "Bob"},
                user="user@test.com",
                server_id="server-1",
                tenant_id="tenant-1",
                plugin_context_table={"existing": True},
                plugin_global_context=global_ctx,
            )

        assert result.messages[0].role == Role.ASSISTANT
        assert result.messages[0].content.text == "post"
        assert global_ctx.user == "user@test.com"
        assert global_ctx.server_id == "server-1"
        assert global_ctx.tenant_id == "tenant-1"

    @pytest.mark.asyncio
    async def test_get_prompt_plugin_hooks_create_global_context_when_missing(self, prompt_service, test_db):
        """Covers GlobalContext creation fallback when middleware didn't provide one."""
        # Standard
        from contextlib import contextmanager

        # First-Party
        from cpex.framework import PromptHookType, PromptPrehookPayload

        db_prompt = _build_db_prompt(template="Hello, {{ name }}!")
        test_db.execute = Mock(return_value=_make_execute_result(scalar=db_prompt))

        plugin_mgr = MagicMock()
        plugin_mgr.has_hooks_for.side_effect = lambda hook: hook == PromptHookType.PROMPT_PRE_FETCH

        pre_payload = PromptPrehookPayload(prompt_id="1", args={"name": "Alice"})
        pre_result = SimpleNamespace(modified_payload=pre_payload)
        plugin_mgr.invoke_hook = AsyncMock(return_value=(pre_result, {"ctx": 1}))

        @contextmanager
        def _no_span(*_a, **_kw):
            yield None

        with (
            patch("mcpgateway.services.prompt_service.create_span", _no_span),
            patch("mcpgateway.services.prompt_service.metrics_buffer") as mock_get_buf,
            patch.object(prompt_service, "_get_plugin_manager", AsyncMock(return_value=plugin_mgr)),
        ):
            mock_get_buf.record_prompt_metric = Mock()
            result = await prompt_service.get_prompt(test_db, "1", {"name": "Bob"}, user="user@test.com")

        assert result.messages[0].content.text == "Hello, Alice!"

    @pytest.mark.asyncio
    async def test_get_prompt_details_not_found(self, prompt_service, test_db):
        test_db.get = Mock(return_value=None)

        with pytest.raises(PromptNotFoundError, match="Prompt not found: 999"):
            await prompt_service.get_prompt_details(test_db, 999)

    @pytest.mark.asyncio
    async def test_get_prompt_details_inactive(self, prompt_service, test_db):
        prompt = _build_db_prompt(is_active=False)
        test_db.get = Mock(return_value=prompt)
        prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1, "enabled": False})

        result = await prompt_service.get_prompt_details(test_db, 1, include_inactive=True)
        assert result["id"] == 1

    # ──────────────────────────────────────────────────────────────────
    #   update_prompt
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_update_prompt_success(self, prompt_service, test_db):
        existing = _build_db_prompt()
        existing.team_id = "team-123"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[  # first call = find existing, second = conflict check
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        test_db.commit = Mock()
        test_db.refresh = Mock()
        prompt_service._notify_prompt_updated = AsyncMock()

        upd = PromptUpdate(description="new desc", template="Hi, {{ name }}!")
        res = await prompt_service.update_prompt(test_db, 1, upd)

        # commit called twice: once for update, once in _get_team_name to release transaction
        assert test_db.commit.call_count == 2
        prompt_service._notify_prompt_updated.assert_called_once()
        assert res["description"] == "new desc"
        assert res["template"] == "Hi, {{ name }}!"

    @pytest.mark.asyncio
    async def test_update_prompt_team_id_rejects_nonexistent_team(self, prompt_service, test_db):
        """Reassigning a prompt to a non-existent team must raise PromptError."""
        existing = _build_db_prompt()
        existing.team_id = "old-team"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(return_value=_make_execute_result(scalar=existing))
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.first.return_value = None  # team not found
        test_db.query = Mock(return_value=mock_query)

        upd = PromptUpdate(team_id="nonexistent-team")

        with pytest.raises(Exception, match="not found"):
            await prompt_service.update_prompt(test_db, 1, upd)

    @pytest.mark.asyncio
    async def test_update_prompt_visibility_team_without_team_id_rejects(self, prompt_service, test_db):
        """Setting visibility to 'team' without any team_id must raise."""
        existing = _build_db_prompt()
        existing.team_id = None
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(return_value=_make_execute_result(scalar=existing))
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.first.return_value = None
        test_db.query = Mock(return_value=mock_query)

        upd = PromptUpdate(visibility="team")

        with pytest.raises(Exception, match="without a team_id"):
            await prompt_service.update_prompt(test_db, 1, upd)

    @pytest.mark.asyncio
    async def test_update_prompt_team_id_rejects_non_owner(self, prompt_service, test_db):
        """Reassigning a prompt to a team where user is not owner must raise."""
        # First-Party
        from mcpgateway.services.prompt_service import _validate_prompt_team_assignment

        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        # Team exists but membership check returns None
        mock_query.first.side_effect = [MagicMock(), None]

        mock_db = MagicMock()
        mock_db.query.return_value = mock_query

        with pytest.raises(ValueError, match="membership"):
            _validate_prompt_team_assignment(mock_db, "user@example.com", "other-team")

    @pytest.mark.asyncio
    async def test_update_prompt_team_id_skips_ownership_without_user_email(self, prompt_service, test_db):
        """System updates without user_email skip ownership checks and persist team_id."""
        existing = _build_db_prompt()
        existing.team_id = "old-team"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        test_db.commit = Mock()
        test_db.refresh = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.first.return_value = MagicMock()  # Team exists
        test_db.query = Mock(return_value=mock_query)
        prompt_service._notify_prompt_updated = AsyncMock()

        upd = PromptUpdate(team_id="new-team")
        await prompt_service.update_prompt(test_db, 1, upd, user_email=None)

        assert existing.team_id == "new-team"

    @pytest.mark.asyncio
    async def test_update_prompt_name_conflict(self, prompt_service, test_db):
        existing = _build_db_prompt()
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        upd = PromptUpdate(name="other")
        with pytest.raises(PromptError):
            await prompt_service.update_prompt(test_db, 1, upd)

    @pytest.mark.asyncio
    async def test_update_prompt_not_found(self, prompt_service, test_db):
        test_db.get = Mock(return_value=None)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=None),  # active
                _make_execute_result(scalar=None),  # inactive
            ]
        )
        upd = PromptUpdate(description="desc")
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.update_prompt(test_db, 999, upd)
        assert "not found" in str(exc_info.value) or "Failed to update prompt" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_update_prompt_inactive(self, prompt_service, test_db):
        inactive = _build_db_prompt(is_active=False)
        test_db.get = Mock(return_value=inactive)
        test_db.commit = Mock()
        test_db.refresh = Mock()
        prompt_service._notify_prompt_updated = AsyncMock()
        upd = PromptUpdate(description="desc")
        res = await prompt_service.update_prompt(test_db, 1, upd)
        assert res["description"] == "desc"

    @pytest.mark.asyncio
    async def test_update_prompt_exception(self, prompt_service, test_db):
        existing = _build_db_prompt()
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(side_effect=[_make_execute_result(scalar=existing), _make_execute_result(scalar=None)])
        test_db.commit = Mock(side_effect=Exception("fail"))
        upd = PromptUpdate(description="desc")
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.update_prompt(test_db, 1, upd)
        assert "Failed to update prompt" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_update_prompt_content_size_error(self, prompt_service, test_db):
        """Test that ContentSizeError is caught and re-raised during prompt update."""
        # First-Party
        from mcpgateway.services.content_security import ContentSizeError

        existing = _build_db_prompt()
        existing.team_id = "team-123"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises ContentSizeError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_size.side_effect = ContentSizeError(content_type="Prompt template", actual_size=15000, max_size=10240)

        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service):
            # Use 15KB template - passes Pydantic (65KB limit) but fails ContentSizeError (10KB limit)
            upd = PromptUpdate(template="x" * 15000)  # 15KB template

            with pytest.raises(ContentSizeError) as exc_info:
                await prompt_service.update_prompt(test_db, 1, upd)

            # Verify the error details
            assert exc_info.value.actual_size == 15000
            assert exc_info.value.max_size == 10240
            assert exc_info.value.content_type == "Prompt template"

            # Verify rollback was called
            test_db.rollback.assert_called_once()

    @pytest.mark.asyncio
    async def test_register_prompt_content_pattern_error(self, prompt_service, test_db, mock_logging_services):
        """Test that ContentPatternError is caught and re-raised during prompt registration."""
        # First-Party
        from mcpgateway.services.content_security import ContentPatternError

        test_db.execute = Mock(return_value=_make_execute_result(scalar=None))
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises ContentPatternError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_size.return_value = None  # Size check passes
        mock_security_service.validate_prompt_template.side_effect = ContentPatternError(
            pattern_matched="__import__", content_type="Prompt template", content_snippet="{{__import__('os')}}", violation_type="python_injection"
        )

        with (
            patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service),
            patch("mcpgateway.services.prompt_service.logger") as mock_logger,
            patch("mcpgateway.services.prompt_service.structured_logger") as mock_structured_logger,
        ):
            # Use model_construct to bypass Pydantic validation
            prompt = PromptCreate.model_construct(name="malicious-prompt", template="{{__import__('os')}}")

            with pytest.raises(ContentPatternError) as exc_info:
                await prompt_service.register_prompt(test_db, prompt, created_by="test_user", owner_email="test@example.com")

            # Verify the error details
            assert exc_info.value.pattern_matched == "__import__"
            assert exc_info.value.violation_type == "python_injection"
            assert exc_info.value.content_snippet == "{{__import__('os')}}"

            # Verify rollback was called
            test_db.rollback.assert_called_once()

            # Verify logger.error was called (covers line 908)
            mock_logger.error.assert_called_once()
            assert "__import__" in str(mock_logger.error.call_args)

            # Verify structured_logger.log was called (covers lines 909-918)
            mock_structured_logger.log.assert_called_once()
            call_args = mock_structured_logger.log.call_args
            assert call_args[1]["level"] == "ERROR"
            assert call_args[1]["message"] == "Prompt creation failed - Malicious pattern detected"
            assert call_args[1]["event_type"] == "prompt_creation_failed"
            assert call_args[1]["component"] == "prompt_service"
            assert call_args[1]["user_id"] == "test_user"
            assert call_args[1]["user_email"] == "test@example.com"
            assert call_args[1]["custom_fields"]["prompt_name"] == "malicious-prompt"

    @pytest.mark.asyncio
    async def test_update_prompt_with_content_pattern_error(self, prompt_service, test_db, mock_logging_services):
        """Test that ContentPatternError is caught and re-raised during prompt update."""
        # First-Party
        from mcpgateway.services.content_security import ContentPatternError

        existing = _build_db_prompt()
        existing.team_id = "team-123"
        test_db.get = Mock(return_value=existing)

        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),  # get_for_update call
                _make_execute_result(scalar=None),  # conflict check (if name changes)
            ]
        )
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises ContentPatternError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_size.return_value = None  # Size check passes
        mock_security_service.validate_prompt_template.side_effect = ContentPatternError(
            pattern_matched="eval(", content_type="Prompt template", content_snippet="{{eval(user_input)}}", violation_type="code_injection"
        )

        mock_structured_logger = mock_logging_services["structured_logger"]

        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service), patch("mcpgateway.services.prompt_service.logger") as mock_logger:
            upd = PromptUpdate(template="{{eval(user_input)}}")

            with pytest.raises(ContentPatternError) as exc_info:
                # Don't pass user_email to skip permission check
                await prompt_service.update_prompt(test_db, 1, upd, modified_by="test_user")

            # Verify the error details
            assert exc_info.value.pattern_matched == "eval("
            assert exc_info.value.violation_type == "code_injection"

            # Verify rollback was called (covers line 2444)
            test_db.rollback.assert_called_once()

            # Verify logger.error was called (covers line 2445)
            mock_logger.error.assert_called_once()
            assert "eval(" in str(mock_logger.error.call_args)

            # Verify structured_logger.log was called (covers line 2446)
            mock_structured_logger.log.assert_called_once()
            call_args = mock_structured_logger.log.call_args
            assert call_args[1]["level"] == "ERROR"
            assert call_args[1]["message"] == "Prompt update failed - Malicious pattern detected"
            assert call_args[1]["event_type"] == "prompt_update_failed"
            assert call_args[1]["component"] == "prompt_service"

    @pytest.mark.asyncio
    async def test_update_prompt_content_pattern_error(self, prompt_service, test_db):
        """Test that TemplateValidationError is caught and re-raised during prompt update.

        This test covers:
        - prompt_service.py template validation with dangerous patterns
        """
        # First-Party
        from mcpgateway.services.content_security import TemplateValidationError

        existing = _build_db_prompt()
        existing.team_id = "team-123"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises TemplateValidationError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_size.return_value = None  # Size check passes
        mock_security_service.validate_prompt_template.side_effect = TemplateValidationError(
            template_name="test-prompt", reason="Template contains dangerous pattern that could lead to code injection", pattern="__import__"
        )

        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service):
            # Use a simple template that passes Pydantic validation
            upd = PromptUpdate(template="Hello {{name}}")

            with pytest.raises(TemplateValidationError) as exc_info:
                await prompt_service.update_prompt(test_db, 1, upd)

            # Verify the error details
            assert exc_info.value.template_name == "test-prompt"
            assert "dangerous pattern" in exc_info.value.reason.lower()

            # Verify rollback was called
            test_db.rollback.assert_called_once()

    @pytest.mark.asyncio
    async def test_update_prompt_template_validation_error(self, prompt_service, test_db):
        """Test that TemplateValidationError is caught and re-raised during prompt update."""
        # First-Party
        from mcpgateway.services.content_security import TemplateValidationError

        existing = _build_db_prompt()
        existing.team_id = "team-123"
        test_db.get = Mock(return_value=existing)
        test_db.execute = Mock(
            side_effect=[
                _make_execute_result(scalar=existing),
                _make_execute_result(scalar=None),
            ]
        )
        test_db.rollback = Mock()

        # Mock get_content_security_service to return a mock that raises TemplateValidationError
        mock_security_service = Mock()
        mock_security_service.validate_prompt_template.side_effect = TemplateValidationError(template_name="test-template", reason="Template contains dangerous pattern", pattern="__import__")

        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service):
            # Use a safe template that passes Pydantic validation
            # The mock will raise TemplateValidationError when validate_prompt_template is called
            upd = PromptUpdate(template="Hello {{ name }}")

            with pytest.raises(TemplateValidationError) as exc_info:
                await prompt_service.update_prompt(test_db, 1, upd)

            # Verify the error details
            assert exc_info.value.template_name == "test-template"
            assert "dangerous pattern" in exc_info.value.reason.lower()
            assert exc_info.value.pattern == "__import__"

            # Verify rollback was called
            test_db.rollback.assert_called_once()

    # ──────────────────────────────────────────────────────────────────
    #   set state
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_set_prompt_state(self, prompt_service, test_db):
        # Ensure the mock prompt has a real id and primitive attributes
        p = MagicMock(spec=DbPrompt)
        p.id = 1
        p.team_id = 1
        p.name = "hello"
        p.is_active = True
        p.enabled = True
        test_db.get = Mock(return_value=p)
        test_db.commit = Mock()
        test_db.refresh = Mock()
        prompt_service._notify_prompt_deactivated = AsyncMock()

        res = await prompt_service.set_prompt_state(test_db, 1, activate=False)

        assert p.enabled is False
        prompt_service._notify_prompt_deactivated.assert_called_once()
        assert res["enabled"] is False

    @pytest.mark.asyncio
    async def test_set_prompt_state_not_found(self, prompt_service, test_db):
        test_db.get = Mock(return_value=None)
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.set_prompt_state(test_db, 999, activate=True)
        assert "Prompt not found" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_set_prompt_state_exception(self, prompt_service, test_db):
        p = _build_db_prompt(is_active=True)
        test_db.get = Mock(return_value=p)
        test_db.commit = Mock(side_effect=Exception("fail"))
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.set_prompt_state(test_db, 1, activate=False)
        assert "Failed to set prompt state" in str(exc_info.value)

    # ──────────────────────────────────────────────────────────────────
    #   delete_prompt
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_delete_prompt_success(self, prompt_service, test_db):
        p = _build_db_prompt()
        test_db.get = Mock(return_value=p)
        test_db.delete = Mock()
        test_db.commit = Mock()
        prompt_service._notify_prompt_deleted = AsyncMock()

        await prompt_service.delete_prompt(test_db, 1)

        test_db.delete.assert_called_once_with(p)
        prompt_service._notify_prompt_deleted.assert_called_once()

    @pytest.mark.asyncio
    async def test_delete_prompt_purge_metrics(self, prompt_service, test_db):
        p = _build_db_prompt()
        test_db.get = Mock(return_value=p)
        test_db.delete = Mock()
        test_db.commit = Mock()
        test_db.execute = Mock()
        prompt_service._notify_prompt_deleted = AsyncMock()

        await prompt_service.delete_prompt(test_db, 1, purge_metrics=True)

        assert test_db.execute.call_count == 2
        test_db.delete.assert_called_once_with(p)
        test_db.commit.assert_called_once()

    @pytest.mark.asyncio
    async def test_delete_prompt_not_found(self, prompt_service, test_db):
        test_db.get = Mock(return_value=None)
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.delete_prompt(test_db, 999)
        assert "Prompt not found" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_delete_prompt_exception(self, prompt_service, test_db):
        p = _build_db_prompt()
        test_db.execute = Mock(return_value=_make_execute_result(scalar=p))
        test_db.delete = Mock(side_effect=Exception("fail"))
        test_db.commit = Mock()
        prompt_service._notify_prompt_deleted = AsyncMock()
        with pytest.raises(PromptError) as exc_info:
            await prompt_service.delete_prompt(test_db, "hello")
        assert "Failed to delete prompt" in str(exc_info.value)

    # ──────────────────────────────────────────────────────────────────
    #   subscribe events logic
    # ──────────────────────────────────────────────────────────────────

    # @pytest.mark.asyncio
    # async def test_subscribe_events_yields_and_unsubscribes(self, prompt_service):
    #     gen = prompt_service.subscribe_events()
    #     # Advance generator to ensure queue is created
    #     await gen.asend(None)
    #     queue = prompt_service._event_subscribers[0]
    #     await queue.put({"type": "test_event"})
    #     event = await gen.__anext__()
    #     assert event["type"] == "test_event"
    #     await gen.aclose()
    #     assert queue not in prompt_service._event_subscribers
    # ──────────────────────────────────────────────────────────────────
    #   Test _publish_event
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_publish_event_puts_in_all_queues(self, prompt_service):
        """Test that _publish_event uses EventService to publish events."""
        # Mock the EventService's publish_event method
        prompt_service._event_service.publish_event = AsyncMock()

        event = {"type": "test"}
        await prompt_service._publish_event(event)

        # Verify that EventService.publish_event was called with the event
        prompt_service._event_service.publish_event.assert_called_once_with(event)

    # ──────────────────────────────────────────────────────────────────
    #   Validation & Exception Handling
    # ──────────────────────────────────────────────────────────────────

    def test_validate_template_raises(self, prompt_service):
        # Patch jinja_env.parse to raise
        prompt_service._jinja_env.parse = Mock(side_effect=Exception("bad"))
        with pytest.raises(PromptValidationError):
            prompt_service._validate_template("bad")

    def test_get_required_arguments(self, prompt_service):
        template = "Hello, {{ name }}! Your code is {{ code }}."
        required = prompt_service._get_required_arguments(template)
        assert "name" in required
        assert "code" in required

    def test_render_template_fallback_and_error(self, prompt_service):
        # Patch _compile_jinja_template to return a template that fails on render
        with patch("mcpgateway.services.prompt_service._compile_jinja_template") as mock_compile:
            mock_template = MagicMock()
            mock_template.render.side_effect = Exception("bad")
            mock_compile.return_value = mock_template

            # Fallback to format
            template = "Hello, {name}!"
            result = prompt_service._render_template(template, {"name": "Alice"})
            assert result == "Hello, Alice!"

            # Format also fails
            with pytest.raises(PromptError):
                prompt_service._render_template(template, {})

    def test_parse_messages_roles(self, prompt_service):
        text = "# User:\nHello\n# Assistant:\nHi!"
        msgs = prompt_service._parse_messages(text)
        assert msgs[0].role == Role.USER
        assert msgs[1].role == Role.ASSISTANT

    def test_parse_messages_starts_with_assistant_and_user_header_flushes_previous(self, prompt_service):
        """Covers branches where a role header appears before any accumulated text."""
        text = "# Assistant:\nHi\n# User:\nHello"
        msgs = prompt_service._parse_messages(text)
        assert len(msgs) == 2
        assert msgs[0].role == Role.ASSISTANT
        assert msgs[1].role == Role.USER

    def test_parse_messages_header_only_returns_empty(self, prompt_service):
        """Covers final 'no current_text' branch."""
        assert prompt_service._parse_messages("# User:") == []

    # ──────────────────────────────────────────────────────────────────
    #   aggregate & reset metrics
    # ──────────────────────────────────────────────────────────────────

    @pytest.mark.asyncio
    async def test_aggregate_and_reset_metrics(self, prompt_service, test_db):
        # Mock aggregate_metrics_combined to return a proper AggregatedMetrics result
        # First-Party
        from mcpgateway.services.metrics_query_service import AggregatedMetrics

        mock_result = AggregatedMetrics(
            total_executions=10,
            successful_executions=8,
            failed_executions=2,
            failure_rate=0.2,
            min_response_time=0.1,
            max_response_time=0.9,
            avg_response_time=0.5,
            last_execution_time="2025-01-01T00:00:00+00:00",
            raw_count=6,
            rollup_count=4,
        )

        with patch("mcpgateway.services.metrics_query_service.aggregate_metrics_combined", return_value=mock_result):
            metrics = await prompt_service.aggregate_metrics(test_db)
            assert isinstance(metrics, PromptMetrics)
            assert metrics.total_executions == 10
            assert metrics.successful_executions == 8
            assert metrics.failed_executions == 2
            assert metrics.failure_rate == 0.2

        # reset_metrics
        test_db.execute = Mock()
        test_db.commit = Mock()
        await prompt_service.reset_metrics(test_db)
        assert test_db.execute.call_count == 2
        test_db.commit.assert_called_once()

    @pytest.mark.asyncio
    async def test_aggregate_metrics_cache_hit_returns_cached(self, prompt_service, test_db):
        PromptMetrics(
            total_executions=123,
            successful_executions=100,
            failed_executions=23,
            failure_rate=0.23,
        )
        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=True),
            patch("mcpgateway.cache.metrics_cache.metrics_cache.get", return_value={"total_executions": 123, "successful_executions": 100, "failed_executions": 23, "failure_rate": 0.23}),
            patch("mcpgateway.services.metrics_query_service.aggregate_metrics_combined") as mock_agg,
        ):
            result = await prompt_service.aggregate_metrics(test_db)
        assert isinstance(result, PromptMetrics)
        mock_agg.assert_not_called()

    @pytest.mark.asyncio
    async def test_aggregate_metrics_cache_disabled_skips_cache_get_set(self, prompt_service, test_db):
        # First-Party
        from mcpgateway.services.metrics_query_service import AggregatedMetrics

        mock_result = AggregatedMetrics(
            total_executions=1,
            successful_executions=1,
            failed_executions=0,
            failure_rate=0.0,
            min_response_time=0.1,
            max_response_time=0.1,
            avg_response_time=0.1,
            last_execution_time="2025-01-01T00:00:00+00:00",
            raw_count=1,
            rollup_count=0,
        )

        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=False),
            patch("mcpgateway.cache.metrics_cache.metrics_cache.get") as mock_get,
            patch("mcpgateway.cache.metrics_cache.metrics_cache.set") as mock_set,
            patch("mcpgateway.services.metrics_query_service.aggregate_metrics_combined", return_value=mock_result),
        ):
            result = await prompt_service.aggregate_metrics(test_db)
        assert result.total_executions == 1
        mock_get.assert_not_called()
        mock_set.assert_not_called()

    @pytest.mark.asyncio
    async def test_list_prompts_with_tags(self, prompt_service, mock_prompt):
        """Test listing prompts with tag filtering."""
        # Third-Party

        # Mock query chain - support pagination methods
        mock_query = MagicMock()
        mock_query.options.return_value = mock_query  # For joinedload
        mock_query.where.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query

        session = MagicMock()
        session.execute.return_value.scalars.return_value.all.return_value = [mock_prompt]

        bind = MagicMock()
        bind.dialect = MagicMock()
        bind.dialect.name = "sqlite"  # or "postgresql"
        session.get_bind.return_value = bind

        with patch("mcpgateway.services.prompt_service.select", return_value=mock_query):
            with patch("mcpgateway.services.prompt_service.json_contains_tag_expr") as mock_json_contains:
                # return a fake condition object that query.where will accept
                fake_condition = MagicMock()
                mock_json_contains.return_value = fake_condition

                result, _ = await prompt_service.list_prompts(session, tags=["test", "production"])

                # helper should be called once with the tags list (not once per tag)
                mock_json_contains.assert_called_once()  # called exactly once
                called_args = mock_json_contains.call_args[0]  # positional args tuple
                assert called_args[0] is session  # session passed through
                # third positional arg is the tags list (signature: session, col, values, match_any=True)
                assert called_args[2] == ["test", "production"]
                # and the fake condition returned must have been passed to where() at some point
                # (there may be multiple where() calls for enabled filter and tags filter)
                mock_query.where.assert_any_call(fake_condition)
                # finally, your service should return the list produced by mock_db.execute(...)
                assert isinstance(result, list)
                assert len(result) == 1


# --------------------------------------------------------------------------- #
#                         Cache Behavior Tests                                #
# --------------------------------------------------------------------------- #


class TestJinjaTemplateCaching:
    """Tests for Jinja template caching (#1814)."""

    def test_template_caching_works(self):
        """Verify template compilation is cached across renders."""
        # First-Party
        from mcpgateway.services.prompt_service import _compile_jinja_template, PromptService

        service = PromptService()
        template = "Hello {{ name }}"

        result1 = service._render_template(template, {"name": "World"})
        assert result1 == "Hello World"

        result2 = service._render_template(template, {"name": "Claude"})
        assert result2 == "Hello Claude"

        info = _compile_jinja_template.cache_info()
        assert info.hits == 1
        assert info.misses == 1

    def test_different_templates_cached_separately(self):
        """Verify different templates get separate cache entries."""
        # First-Party
        from mcpgateway.services.prompt_service import _compile_jinja_template, PromptService

        service = PromptService()

        result1 = service._render_template("Hello {{ name }}", {"name": "A"})
        result2 = service._render_template("Goodbye {{ name }}", {"name": "B"})

        assert result1 == "Hello A"
        assert result2 == "Goodbye B"

        info = _compile_jinja_template.cache_info()
        assert info.misses == 2  # Two different templates

    def test_format_fallback_still_works(self):
        """Verify Python format() fallback works when Jinja render fails."""
        # First-Party
        from mcpgateway.services.prompt_service import PromptService

        service = PromptService()

        # Mock _compile_jinja_template to return a template that fails on render
        with patch("mcpgateway.services.prompt_service._compile_jinja_template") as mock_compile:
            mock_template = MagicMock()
            mock_template.render.side_effect = Exception("Jinja render error")
            mock_compile.return_value = mock_template

            # Should fall back to Python format()
            template = "Hello, {name}!"
            result = service._render_template(template, {"name": "Alice"})
            assert result == "Hello, Alice!"


class TestPromptAccessAuthorization:
    """Tests for _check_prompt_access authorization logic."""

    @pytest.fixture
    def prompt_service(self):
        """Create a prompt service instance."""
        return PromptService()

    @pytest.fixture
    def mock_db(self):
        """Create a mock database session."""
        db = MagicMock()
        db.commit = Mock()
        return db

    def _create_mock_prompt(self, visibility="public", owner_email=None, team_id=None):
        """Helper to create mock prompt."""
        prompt = MagicMock()
        prompt.visibility = visibility
        prompt.owner_email = owner_email
        prompt.team_id = team_id
        return prompt

    @pytest.mark.asyncio
    async def test_check_prompt_access_public_always_allowed(self, prompt_service, mock_db):
        """Public prompts should be accessible to anyone."""
        public_prompt = self._create_mock_prompt(visibility="public")

        # Unauthenticated
        assert await prompt_service._check_prompt_access(mock_db, public_prompt, user_email=None, token_teams=[]) is True
        # Authenticated
        assert await prompt_service._check_prompt_access(mock_db, public_prompt, user_email="user@test.com", token_teams=["team-1"]) is True
        # Admin
        assert await prompt_service._check_prompt_access(mock_db, public_prompt, user_email=None, token_teams=None) is True

    @pytest.mark.asyncio
    async def test_check_prompt_access_admin_bypass_denied_for_private(self, prompt_service, mock_db):
        """Admin bypass does NOT grant access to private resources (security requirement)."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="secret@test.com", team_id="secret-team")

        # Admin bypass: both None, but private resources are NEVER accessible via admin bypass
        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email=None, token_teams=None) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_admin_bypass_grants_team_access(self, prompt_service, mock_db):
        """Admin bypass grants access to team resources."""
        team_prompt = self._create_mock_prompt(visibility="team", owner_email="owner@test.com", team_id="team-abc")

        # Admin bypass: both None = access to team resources
        assert await prompt_service._check_prompt_access(mock_db, team_prompt, user_email=None, token_teams=None) is True

    @pytest.mark.asyncio
    async def test_check_prompt_access_database_admin_bypass(self, prompt_service, mock_db):
        """DB admin bypass: own private allowed, other user's private denied (PR #4341)."""
        other_users_private = self._create_mock_prompt(visibility="private", owner_email="secret@test.com", team_id="secret-team")
        own_private = self._create_mock_prompt(visibility="private", owner_email="admin@test.com", team_id="secret-team")

        install_admin_user(mock_db)

        # token_teams=None + DB admin viewing OWN private → allowed (#4341 carve-out for self-access)
        assert await prompt_service._check_prompt_access(mock_db, own_private, user_email="admin@test.com", token_teams=None) is True
        # token_teams=None + DB admin viewing OTHER user's private → denied (#4341 invariant)
        assert await prompt_service._check_prompt_access(mock_db, other_users_private, user_email="admin@test.com", token_teams=None) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_admin_with_narrowed_token_still_narrowed(self, prompt_service, mock_db):
        """DB admin with a team-scoped token must NOT bypass (#4106 guard)."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="secret@test.com", team_id="secret-team")

        install_admin_user(mock_db)

        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email="admin@test.com", token_teams=["some-team"]) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_admin_with_public_only_token_stays_public_only(self, prompt_service, mock_db):
        """DB admin with public-only token (token_teams=[]) sees only public."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="secret@test.com", team_id="secret-team")

        install_admin_user(mock_db)

        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email="admin@test.com", token_teams=[]) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_private_denied_to_unauthenticated(self, prompt_service, mock_db):
        """Private prompts should be denied to unauthenticated users."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="owner@test.com")

        # Unauthenticated (public-only token)
        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email=None, token_teams=[]) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_private_allowed_to_owner(self, prompt_service, mock_db):
        """Private prompts should be accessible to the owner."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="owner@test.com")

        # Owner with non-empty token_teams
        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email="owner@test.com", token_teams=["some-team"]) is True

    @pytest.mark.asyncio
    async def test_check_prompt_access_private_denied_to_non_owner_when_authenticated(self, prompt_service, mock_db):
        """Private prompts should be denied to authenticated non-owners (covers prompt_team_id=None path)."""
        private_prompt = self._create_mock_prompt(visibility="private", owner_email="owner@test.com", team_id=None)

        assert await prompt_service._check_prompt_access(mock_db, private_prompt, user_email="other@test.com", token_teams=["team-1"]) is False

    @pytest.mark.asyncio
    async def test_check_prompt_access_team_prompt_allowed_to_member(self, prompt_service, mock_db):
        """Team prompts should be accessible to team members."""
        team_prompt = self._create_mock_prompt(visibility="team", owner_email="owner@test.com", team_id="team-abc")

        # Team member via token_teams
        assert await prompt_service._check_prompt_access(mock_db, team_prompt, user_email="member@test.com", token_teams=["team-abc"]) is True

    @pytest.mark.asyncio
    async def test_check_prompt_access_team_prompt_db_lookup_allows_member(self, prompt_service, mock_db):
        """Team prompts should consult DB when token_teams is None."""
        team_prompt = self._create_mock_prompt(visibility="team", owner_email="owner@test.com", team_id="team-abc")

        team = MagicMock()
        team.id = "team-abc"

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[team])
            MockTMS.return_value = mock_ts

            assert await prompt_service._check_prompt_access(mock_db, team_prompt, user_email="member@test.com", token_teams=None) is True

    @pytest.mark.asyncio
    async def test_check_prompt_access_team_prompt_denied_to_non_member(self, prompt_service, mock_db):
        """Team prompts should be denied to non-members."""
        team_prompt = self._create_mock_prompt(visibility="team", owner_email="owner@test.com", team_id="team-abc")

        # Non-member
        assert await prompt_service._check_prompt_access(mock_db, team_prompt, user_email="outsider@test.com", token_teams=["other-team"]) is False


# --------------------------------------------------------------------------- #
# Prompt Namespacing tests                                                    #
# --------------------------------------------------------------------------- #


class TestPromptGatewayNamespacing:
    """Test prompt namespacing by gateway_id."""

    @pytest.mark.asyncio
    async def test_prompt_namespacing_different_gateways(self, prompt_service, test_db):
        """Test: Same `name` can be registered for **different** gateways (same team/owner).

        Verifies that the conflict query includes gateway_id in the filter by capturing
        the executed SQL and checking for the gateway_id clause.
        """
        # First-Party

        # Setup prompt create data
        pc = PromptCreate(name="hello", description="greet a user", template="Hello {{ name }}!", arguments=[], gateway_id="gateway-2")

        # Track executed queries to verify gateway_id filtering
        executed_queries = []

        def capture_execute(stmt):
            executed_queries.append(str(stmt))
            # First call: gateway lookup (returns None - no gateway found)
            # Second call: conflict check (returns None - no conflict)
            return _make_execute_result(scalar=None)

        test_db.execute = Mock(side_effect=capture_execute)
        test_db.add, test_db.commit, test_db.refresh = Mock(), Mock(), Mock()

        prompt_service._notify_prompt_added = AsyncMock()

        # Execution
        _ = await prompt_service.register_prompt(test_db, pc)

        # Verification: check that gateway_id was included in the conflict query
        test_db.add.assert_called_once()
        stmt = test_db.add.call_args[0][0]
        assert stmt.name == "hello"
        assert stmt.gateway_id == "gateway-2"

        # Verify the conflict check query included gateway_id
        # The second query should be the conflict check
        assert len(executed_queries) >= 2, "Expected at least 2 queries (gateway lookup + conflict check)"
        conflict_query = executed_queries[1]
        assert "gateway_id" in conflict_query, f"Conflict query must filter by gateway_id: {conflict_query}"

    @pytest.mark.asyncio
    async def test_prompt_namespacing_same_gateway(self, prompt_service, test_db):
        """Test: Same `name` **cannot** be registered for the **same** gateway (same team/owner)."""
        # First-Party

        # Setup existing prompt
        existing = _build_db_prompt(name="hello")
        existing.gateway_id = "gateway-1"
        existing.visibility = "public"

        call_count = [0]

        def mock_execute(stmt):
            call_count[0] += 1
            query_str = str(stmt)
            if call_count[0] == 1:
                # First call: gateway lookup - return None
                return _make_execute_result(scalar=None)
            # Second call: conflict check - return existing prompt
            # Verify gateway_id is in the query
            assert "gateway_id" in query_str, f"Conflict query must include gateway_id: {query_str}"
            return _make_execute_result(scalar=existing)

        test_db.execute = Mock(side_effect=mock_execute)

        pc = PromptCreate(name="hello", description="", template="X", arguments=[], gateway_id="gateway-1")

        with pytest.raises(PromptError) as exc_info:
            await prompt_service.register_prompt(test_db, pc)

        assert "already exists" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_prompt_namespacing_local_prompts(self, prompt_service, test_db):
        """Test: Local prompts (`gateway_id=NULL`) still enforce uniqueness per team/owner."""
        # Setup existing local prompt
        existing = _build_db_prompt(name="hello")
        existing.gateway_id = None
        existing.visibility = "public"

        # Track executed queries to verify gateway_id filtering
        executed_queries = []

        def mock_execute(stmt):
            query_str = str(stmt)
            executed_queries.append(query_str)
            # When gateway_id=None, no gateway lookup occurs - first call is conflict check
            # Return existing prompt to trigger conflict error
            return _make_execute_result(scalar=existing)

        test_db.execute = Mock(side_effect=mock_execute)

        pc = PromptCreate(name="hello", description="", template="X", arguments=[], gateway_id=None)

        with pytest.raises(PromptError) as exc_info:
            await prompt_service.register_prompt(test_db, pc)

        assert "already exists" in str(exc_info.value)

        # Verify the conflict check query included gateway_id
        assert len(executed_queries) >= 1, "Expected at least 1 query (conflict check)"
        conflict_query = executed_queries[0]
        assert "gateway_id" in conflict_query, f"Conflict query must include gateway_id: {conflict_query}"


class TestPromptBulkRegistration:
    """Additional coverage for bulk prompt registration branches."""

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_empty_returns_zeroes(self, prompt_service):
        result = await prompt_service.register_prompts_bulk(db=MagicMock(), prompts=[])

        assert result == {"created": 0, "updated": 0, "skipped": 0, "failed": 0, "errors": []}

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_update_conflict_updates_existing(self, prompt_service):
        existing = MagicMock(spec=DbPrompt)
        existing.name = prompt_service._compute_prompt_name("custom")
        existing.gateway_id = None
        existing.description = "Old"
        existing.template = "Old {{ name }}"
        existing.argument_schema = {}
        existing.tags = ["old"]
        existing.custom_name = "old"
        existing.display_name = "old"
        existing.version = 1

        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = [existing]
        db.add_all = MagicMock()
        db.commit = MagicMock()
        db.refresh = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="prompt",
            custom_name="custom",
            display_name="display",
            description="New desc",
            template="Hello {{ name }}",
            arguments=[PromptArgument(name="name", description="who")],
            tags=["new"],
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="update",
        )

        assert result["updated"] == 1
        assert existing.description == "New desc"
        assert existing.template == "Hello {{ name }}"
        assert existing.tags[0]["id"] == "new"
        assert existing.tags[0]["label"] == "new"
        assert existing.custom_name == "custom"
        assert existing.display_name == "display"
        assert existing.version == 2
        assert existing.argument_schema["properties"]["name"]["description"] == "who"
        db.add_all.assert_not_called()

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_update_conflict_skips_custom_and_display_when_missing(self, prompt_service):
        """Covers 'custom_name/display_name is None' branches during bulk update."""
        existing = MagicMock(spec=DbPrompt)
        existing.name = prompt_service._compute_prompt_name("prompt")
        existing.gateway_id = None
        existing.description = "Old"
        existing.template = "Old {{ name }}"
        existing.argument_schema = {}
        existing.tags = []
        existing.custom_name = "keep-custom"
        existing.display_name = "keep-display"
        existing.version = 1

        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = [existing]
        db.add_all = MagicMock()
        db.commit = MagicMock()
        db.refresh = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="prompt",
            description="New desc",
            template="Hello {{ name }}",
            arguments=[PromptArgument(name="name")],
            tags=["new"],
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="update",
        )

        assert result["updated"] == 1
        assert existing.custom_name == "keep-custom"
        assert existing.display_name == "keep-display"

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_rename_conflict_with_gateway(self, prompt_service):
        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.name = "Gateway One"

        computed_name = prompt_service._compute_prompt_name("conflict", gateway=gateway)
        existing = MagicMock(spec=DbPrompt)
        existing.name = computed_name
        existing.gateway_id = "gw-1"

        gateway_result = MagicMock()
        gateway_result.scalars.return_value.all.return_value = [gateway]
        prompts_result = MagicMock()
        prompts_result.scalars.return_value.all.return_value = [existing]

        db = MagicMock()
        db.execute.side_effect = [gateway_result, prompts_result]
        db.add_all = MagicMock()
        db.commit = MagicMock()
        db.refresh = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="conflict",
            template="Hello {{ name }}",
            arguments=[],
            gateway_id="gw-1",
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="rename",
            visibility="team",
            team_id="team-1",
        )

        assert result["created"] == 1
        added = db.add_all.call_args.args[0][0]
        assert added.custom_name.startswith("conflict_imported_")
        assert added.display_name.startswith("conflict_imported_")
        assert added.gateway is gateway
        assert added.gateway_name_cache == "Gateway One"
        assert added.team_id == "team-1"
        assert added.visibility == "team"

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_creates_new_prompt_with_gateway_sets_relationship(self, prompt_service):
        """Covers new prompt creation branch with gateway relationship assignment."""
        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.name = "Gateway One"

        gateway_result = MagicMock()
        gateway_result.scalars.return_value.all.return_value = [gateway]
        prompts_result = MagicMock()
        prompts_result.scalars.return_value.all.return_value = []

        db = MagicMock()
        db.execute.side_effect = [gateway_result, prompts_result]
        db.add_all = MagicMock()
        db.commit = MagicMock()
        db.refresh = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="new",
            template="Hello!",
            arguments=[],
            gateway_id="gw-1",
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="skip",
        )

        assert result["created"] == 1
        added = db.add_all.call_args.args[0][0]
        assert added.gateway is gateway
        assert added.gateway_name_cache == "Gateway One"

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_fail_conflict_records_error(self, prompt_service):
        existing = MagicMock(spec=DbPrompt)
        existing.name = "conflict"
        existing.gateway_id = None

        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = [existing]
        db.commit = MagicMock()
        db.refresh = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="conflict",
            template="Hello {{ name }}",
            arguments=[],
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="fail",
            visibility="private",
            owner_email="owner@example.com",
        )

        assert result["failed"] == 1
        assert any("Prompt name conflict" in err for err in result["errors"])

    @pytest.mark.asyncio
    async def test_register_prompts_bulk_invalid_template_counts_failed(self, prompt_service, monkeypatch):
        """Test that template validation errors cause fail-fast in bulk operations."""
        # First-Party
        from mcpgateway import config
        from mcpgateway.services.content_security import ContentSecurityService, TemplateValidationError

        # Ensure validation is enabled by monkeypatching settings
        monkeypatch.setattr(config.settings, "content_validate_prompt_templates", True)

        # Create a fresh ContentSecurityService with validation enabled
        mock_security_service = ContentSecurityService()

        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = []
        db.commit = MagicMock()
        db.refresh = MagicMock()
        db.rollback = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = SimpleNamespace(
            name="bad",
            template="Hello {{ invalid",  # Unbalanced braces
            description=None,
            arguments=[],
            tags=[],
            custom_name=None,
            display_name=None,
            gateway_id=None,
            team_id=None,
            owner_email=None,
            visibility="public",
        )

        # Mock get_content_security_service to return our service with validation enabled
        with patch("mcpgateway.services.prompt_service.get_content_security_service", return_value=mock_security_service):
            # Template validation errors now cause fail-fast behavior
            with pytest.raises(TemplateValidationError) as exc_info:
                await prompt_service.register_prompts_bulk(
                    db=db,
                    prompts=[prompt],
                    created_by="tester",
                    conflict_strategy="skip",
                )

            assert "Unbalanced template braces" in str(exc_info.value)
            # Verify rollback was called
            db.rollback.assert_called()


# ---------------------------------------------------------------------------
# Additional coverage tests
# ---------------------------------------------------------------------------


class TestGetTopPrompts:
    """Tests for get_top_prompts (lines 236-287)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_cache_hit(self, prompt_service):
        db = MagicMock()
        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=True),
            patch("mcpgateway.cache.metrics_cache.metrics_cache") as mock_cache,
        ):
            mock_cache.get.return_value = [{"id": 1, "name": "cached"}]
            result = await prompt_service.get_top_prompts(db)
        assert result == [{"id": 1, "name": "cached"}]

    @pytest.mark.asyncio
    async def test_cache_miss_queries_db(self, prompt_service):
        db = MagicMock()
        mock_results = MagicMock()
        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=True),
            patch("mcpgateway.cache.metrics_cache.metrics_cache") as mock_cache,
            patch("mcpgateway.services.metrics_query_service.get_top_performers_combined", return_value=mock_results),
            patch("mcpgateway.services.prompt_service.build_top_performers", return_value=["top1"]),
        ):
            mock_cache.get.return_value = None
            result = await prompt_service.get_top_prompts(db, limit=3)
        assert result == ["top1"]
        mock_cache.set.assert_called_once()

    @pytest.mark.asyncio
    async def test_cache_disabled(self, prompt_service):
        db = MagicMock()
        mock_results = MagicMock()
        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=False),
            patch("mcpgateway.services.metrics_query_service.get_top_performers_combined", return_value=mock_results),
            patch("mcpgateway.services.prompt_service.build_top_performers", return_value=["top1"]),
        ):
            result = await prompt_service.get_top_prompts(db)
        assert result == ["top1"]

    @pytest.mark.asyncio
    async def test_include_deleted(self, prompt_service):
        db = MagicMock()
        mock_results = MagicMock()
        with (
            patch("mcpgateway.cache.metrics_cache.is_cache_enabled", return_value=False),
            patch("mcpgateway.services.metrics_query_service.get_top_performers_combined", return_value=mock_results) as mock_gtp,
            patch("mcpgateway.services.prompt_service.build_top_performers", return_value=[]),
        ):
            await prompt_service.get_top_prompts(db, include_deleted=True)
            assert mock_gtp.call_args[1]["include_deleted"] is True


class TestConvertPromptToRead:
    """Tests for convert_prompt_to_read (lines 289-375)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    def test_without_metrics(self, prompt_service):
        p = _build_db_prompt(pid=10, name="my-prompt", desc="A prompt")
        result = prompt_service.convert_prompt_to_read(p, include_metrics=False)
        assert result["id"] == 10
        assert result["name"] == "my-prompt"
        assert result["metrics"] is None

    def test_with_metrics(self, prompt_service):
        m1 = MagicMock()
        m1.is_success = True
        m1.response_time = 0.5
        m1.timestamp = datetime(2025, 6, 1, tzinfo=timezone.utc)
        m2 = MagicMock()
        m2.is_success = False
        m2.response_time = 1.0
        m2.timestamp = datetime(2025, 6, 2, tzinfo=timezone.utc)

        p = _build_db_prompt(pid=11, metrics=[m1, m2])
        result = prompt_service.convert_prompt_to_read(p, include_metrics=True)
        assert result["metrics"]["totalExecutions"] == 2
        assert result["metrics"]["successfulExecutions"] == 1
        assert result["metrics"]["failedExecutions"] == 1
        assert result["metrics"]["avgResponseTime"] == 0.75
        assert result["metrics"]["minResponseTime"] == 0.5
        assert result["metrics"]["maxResponseTime"] == 1.0

    def test_with_empty_metrics(self, prompt_service):
        p = _build_db_prompt(pid=12, metrics=[])
        result = prompt_service.convert_prompt_to_read(p, include_metrics=True)
        assert result["metrics"]["totalExecutions"] == 0
        assert result["metrics"]["avgResponseTime"] is None

    def test_arguments_from_schema(self, prompt_service):
        p = _build_db_prompt()
        p.argument_schema = {"properties": {"name": {"type": "string", "description": "User name"}}, "required": ["name"]}
        result = prompt_service.convert_prompt_to_read(p)
        assert len(result["arguments"]) == 1
        assert result["arguments"][0]["name"] == "name"
        assert result["arguments"][0]["required"] is True


class TestGetTeamName:
    """Tests for _get_team_name (lines 377-391)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    def test_none_team_id(self, prompt_service):
        db = MagicMock()
        assert prompt_service._get_team_name(db, None) is None

    def test_team_found(self, prompt_service):
        db = MagicMock()
        mock_team = MagicMock()
        mock_team.name = "Engineering"
        db.query.return_value.filter.return_value.first.return_value = mock_team
        assert prompt_service._get_team_name(db, "team-1") == "Engineering"

    def test_team_not_found(self, prompt_service):
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = None
        assert prompt_service._get_team_name(db, "team-99") is None


class TestComputePromptName:
    """Tests for _compute_prompt_name (lines 393-407)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    def test_no_gateway(self, prompt_service):
        result = prompt_service._compute_prompt_name("My Prompt")
        assert result == "my-prompt"

    def test_with_gateway(self, prompt_service, monkeypatch):
        monkeypatch.setattr("mcpgateway.services.prompt_service.settings", MagicMock(gateway_tool_name_separator="__"))
        gateway = MagicMock()
        gateway.name = "Test Gateway"
        result = prompt_service._compute_prompt_name("My Prompt", gateway=gateway)
        assert result == "test-gateway__my-prompt"


class TestListPromptsAdvanced:
    """Tests for list_prompts pagination and filtering (lines 969-1152)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_page_based_pagination(self, prompt_service):
        db = MagicMock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = None

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache"),
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_paginate.return_value = {
                "data": [mock_prompt],
                "pagination": {"page": 1, "per_page": 10, "total": 1},
                "links": {"self": "/admin/prompts?page=1"},
            }

            result = await prompt_service.list_prompts(db, page=1, per_page=10)

        assert "data" in result
        assert "pagination" in result
        assert result["data"] == ["converted"]

    @pytest.mark.asyncio
    async def test_cache_hit(self, prompt_service):
        db = MagicMock()
        with patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn:
            mock_cache = AsyncMock()
            mock_cache.hash_filters.return_value = "hash"
            mock_cache.get = AsyncMock(return_value={"prompts": [{"id": 1, "name": "cached"}], "next_cursor": None})
            mock_cache_fn.return_value = mock_cache

            result, cursor = await prompt_service.list_prompts(db)
        assert len(result) == 1
        assert cursor is None

    @pytest.mark.asyncio
    async def test_token_teams_empty_public_only(self, prompt_service):
        """Empty token_teams should only show public prompts."""
        db = MagicMock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = None

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_paginate.return_value = ([mock_prompt], None)

            result, cursor = await prompt_service.list_prompts(db, token_teams=[])
        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_token_teams_with_teams(self, prompt_service):
        db = MagicMock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = "team-1"

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_paginate.return_value = ([mock_prompt], None)

            result, cursor = await prompt_service.list_prompts(db, token_teams=["team-1"])
        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_user_email_with_team_id_no_access(self, prompt_service):
        db = MagicMock()

        with (
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result, cursor = await prompt_service.list_prompts(db, user_email="user@test.com", team_id="team-99")
        assert result == []

    @pytest.mark.asyncio
    async def test_user_email_empty_string_treated_as_public_only(self, prompt_service):
        """An empty (but present) user_email should still apply secure public-only filtering."""
        db = MagicMock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = None

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_paginate.return_value = ([mock_prompt], None)

            result, _ = await prompt_service.list_prompts(db, user_email="")

        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_token_teams_with_team_id_access_and_visibility_filter(self, prompt_service):
        """Covers team_id access conditions + visibility filter branch."""
        db = MagicMock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = "team-1"

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_paginate.return_value = ([mock_prompt], None)

            result, _ = await prompt_service.list_prompts(
                db,
                user_email="user@test.com",
                token_teams=["team-1"],
                team_id="team-1",
                visibility="team",
            )

        assert result == ["converted"]


class TestListPromptsForUser:
    """Tests for list_prompts_for_user (lines 1154-1243)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_basic_listing(self, prompt_service):
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = None
        db.execute.return_value.scalars.return_value.all.return_value = [mock_prompt]

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com")

        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_team_no_access(self, prompt_service):
        db = MagicMock()
        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com", team_id="team-99")
        assert result == []

    @pytest.mark.asyncio
    async def test_team_with_access(self, prompt_service):
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = "team-1"

        # Use side_effect to return different results for sequential db.execute() calls
        main_result = MagicMock()
        main_result.scalars.return_value.all.return_value = [mock_prompt]
        team_result = MagicMock()
        team_result.all.return_value = [MagicMock(id="team-1", name="Team")]
        db.execute = MagicMock(side_effect=[main_result, team_result])

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        team = MagicMock()
        team.id = "team-1"
        team.name = "Team"

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[team])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com", team_id="team-1")

        assert len(result) == 1

    @pytest.mark.asyncio
    async def test_conversion_error_skipped(self, prompt_service):
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = None
        db.execute.return_value.scalars.return_value.all.return_value = [mock_prompt]

        prompt_service.convert_prompt_to_read = MagicMock(side_effect=ValueError("bad"))

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com")

        assert result == []

    @pytest.mark.asyncio
    async def test_visibility_filter(self, prompt_service):
        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = []

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com", visibility="private")

        assert result == []

    @pytest.mark.asyncio
    async def test_include_inactive(self, prompt_service):
        db = MagicMock()
        db.execute.return_value.scalars.return_value.all.return_value = []

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_prompts_for_user(db, "user@test.com", include_inactive=True)

        assert result == []


class TestRecordPromptMetric:
    """Tests for _record_prompt_metric (lines 1348-1370)."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_success(self, prompt_service):
        db = MagicMock()
        prompt = _build_db_prompt(pid="prompt-1")
        # Standard
        import time

        start = time.monotonic() - 0.5
        await prompt_service._record_prompt_metric(db, prompt, start, True, None)
        db.add.assert_called_once()
        metric = db.add.call_args[0][0]
        assert metric.prompt_id == "prompt-1"
        assert metric.is_success is True
        assert metric.response_time > 0
        db.commit.assert_called_once()

    @pytest.mark.asyncio
    async def test_failure(self, prompt_service):
        db = MagicMock()
        prompt = _build_db_prompt(pid="prompt-2")
        # Standard
        import time

        start = time.monotonic()
        await prompt_service._record_prompt_metric(db, prompt, start, False, "error msg")
        metric = db.add.call_args[0][0]
        assert metric.is_success is False
        assert metric.error_message == "error msg"


# ---------------------------------------------------------------------------
#  Additional coverage: list_server_prompts
# ---------------------------------------------------------------------------


class TestListServerPrompts:
    """Cover lines 1289-1346: list_server_prompts server association + team batch fetch."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_basic_listing(self, prompt_service):
        """Basic listing returns prompts for a server."""
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = None

        # First execute: prompt query
        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [mock_prompt]
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        result = await prompt_service.list_server_prompts(db, "server-1")
        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_list_server_prompts_with_include_metrics_true(self, prompt_service):
        """Test that list_server_prompts eager loads metrics when include_metrics=True.

        This test ensures that when include_metrics=True, the query includes
        selectinload for both metrics and metrics_hourly relationships to prevent N+1 queries.
        Regression test for PR #3649 performance optimization.
        """
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.enabled = True
        mock_prompt.team_id = None
        mock_prompt.team = None

        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [mock_prompt]
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted_prompt_with_metrics")

        # Call with include_metrics=True to trigger eager loading code path
        prompts = await prompt_service.list_server_prompts(db, server_id="server-1", include_metrics=True)

        assert prompts == ["converted_prompt_with_metrics"]
        # Verify convert_prompt_to_read was called with include_metrics=True
        prompt_service.convert_prompt_to_read.assert_called_once_with(mock_prompt, include_metrics=True)

    @pytest.mark.asyncio
    async def test_include_inactive_true_skips_enabled_filter(self, prompt_service):
        """Covers include_inactive=True branch (skips DbPrompt.enabled filter)."""
        db = MagicMock()
        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = []
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        result = await prompt_service.list_server_prompts(db, "server-1", include_inactive=True)
        assert result == []

    @pytest.mark.asyncio
    async def test_user_email_db_team_lookup_path(self, prompt_service):
        """Covers TeamManagementService DB lookup path when token_teams is None."""
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = None

        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [mock_prompt]
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        team = MagicMock()
        team.id = "team-1"

        with patch("mcpgateway.services.prompt_service.TeamManagementService") as MockTMS:
            mock_ts = MagicMock()
            mock_ts.get_user_teams = AsyncMock(return_value=[team])
            MockTMS.return_value = mock_ts

            result = await prompt_service.list_server_prompts(db, "server-1", user_email="user@test.com")

        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_user_email_empty_string_sets_team_ids_empty(self, prompt_service):
        """Empty string user_email is treated as provided but unauthenticated (public-only)."""
        db = MagicMock()
        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = []
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        result = await prompt_service.list_server_prompts(db, "server-1", user_email="")
        assert result == []

    @pytest.mark.asyncio
    async def test_with_team_batch_fetch(self, prompt_service):
        """Prompts with team_ids trigger batch team name fetch."""
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = "team-1"

        # First call: prompt query; second call: team name batch query
        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [mock_prompt]
        team_result = MagicMock()
        team_row = MagicMock()
        team_row.id = "team-1"
        team_row.name = "Engineering"
        team_result.all.return_value = [team_row]
        db.execute = MagicMock(side_effect=[prompt_result, team_result])
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        result = await prompt_service.list_server_prompts(db, "server-1")
        assert result == ["converted"]
        # team was set on the prompt
        assert mock_prompt.team == "Engineering"

    @pytest.mark.asyncio
    async def test_with_token_teams_public_only(self, prompt_service):
        """token_teams=[] restricts to public-only prompts."""
        db = MagicMock()
        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = []
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        result = await prompt_service.list_server_prompts(db, "server-1", token_teams=[])
        assert result == []

    @pytest.mark.asyncio
    async def test_with_token_teams_scoped(self, prompt_service):
        """token_teams=["team-1"] shows public + team prompts."""
        db = MagicMock()
        mock_prompt = MagicMock()
        mock_prompt.team_id = "team-1"

        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [mock_prompt]
        team_result = MagicMock()
        team_result.all.return_value = []
        db.execute = MagicMock(side_effect=[prompt_result, team_result])
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(return_value="converted")

        result = await prompt_service.list_server_prompts(db, "server-1", token_teams=["team-1"], user_email="user@test.com")
        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_conversion_error_skipped(self, prompt_service):
        """Conversion errors for individual prompts don't fail the whole list."""
        db = MagicMock()
        p1 = MagicMock()
        p1.team_id = None
        p2 = MagicMock()
        p2.team_id = None

        prompt_result = MagicMock()
        prompt_result.scalars.return_value.all.return_value = [p1, p2]
        db.execute = MagicMock(return_value=prompt_result)
        db.commit = MagicMock()

        prompt_service.convert_prompt_to_read = MagicMock(side_effect=[ValueError("bad"), "ok"])

        result = await prompt_service.list_server_prompts(db, "server-1")
        assert result == ["ok"]


# ---------------------------------------------------------------------------
#  Additional coverage: update_prompt name conflict detection
# ---------------------------------------------------------------------------


class TestUpdatePromptNameConflict:
    """Cover lines 1800-1810: team/private name conflict in update_prompt."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_team_name_conflict(self, prompt_service):
        """Name conflict in team visibility raises PromptError."""
        existing = _build_db_prompt(name="old-name")
        existing.visibility = "team"
        existing.team_id = "team-1"
        existing.custom_name = "old-name"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"

        conflicting = MagicMock()
        conflicting.enabled = True
        conflicting.id = 99
        conflicting.visibility = "team"

        with (
            patch("mcpgateway.services.prompt_service.get_for_update") as mock_gfu,
            patch("mcpgateway.services.prompt_service._get_registry_cache"),
        ):
            mock_gfu.side_effect = [existing, conflicting]  # first: get prompt, second: conflict check

            upd = PromptUpdate(name="new-name")

            with pytest.raises(PromptError):
                await prompt_service.update_prompt(MagicMock(), 1, upd)

    @pytest.mark.asyncio
    async def test_private_name_conflict(self, prompt_service):
        """Name conflict in private visibility raises PromptError."""
        existing = _build_db_prompt(name="old-name")
        existing.visibility = "private"
        existing.team_id = None
        existing.custom_name = "old-name"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"

        conflicting = MagicMock()
        conflicting.enabled = True
        conflicting.id = 99
        conflicting.visibility = "private"

        with (patch("mcpgateway.services.prompt_service.get_for_update") as mock_gfu,):
            mock_gfu.side_effect = [existing, conflicting]

            upd = PromptUpdate(name="new-name")

            with pytest.raises(PromptError):
                await prompt_service.update_prompt(MagicMock(), 1, upd)


# ---------------------------------------------------------------------------
#  Additional coverage: update_prompt field updates and exception handlers
# ---------------------------------------------------------------------------


class TestUpdatePromptFieldsAndExceptions:
    """Cover lines 1822-1873, 1924-1953."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_update_with_arguments_and_version(self, prompt_service):
        """Updating template+arguments regenerates argument schema and increments version."""
        existing = _build_db_prompt(name="my-prompt")
        existing.visibility = "public"
        existing.team_id = "t1"
        existing.custom_name = "my-prompt"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"
        existing.version = 3

        db = MagicMock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", return_value=existing),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            db.commit = Mock()
            db.refresh = Mock()
            prompt_service._notify_prompt_updated = AsyncMock()
            prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1})

            upd = PromptUpdate(
                template="Hi {{ user }}!",
                arguments=[PromptArgument(name="user", description="Username")],
                visibility="public",
                tags=["v2"],
            )

            await prompt_service.update_prompt(db, 1, upd, modified_by="admin", modified_from_ip="1.2.3.4", modified_via="api", modified_user_agent="test-agent")

        assert existing.template == "Hi {{ user }}!"
        assert existing.argument_schema["properties"]["user"]["description"] == "Username"
        assert existing.version == 4
        assert existing.tags is not None
        assert existing.visibility == "public"

    @pytest.mark.asyncio
    async def test_update_permission_denied(self, prompt_service):
        """Permission error during update is propagated."""
        existing = _build_db_prompt(name="owned")
        existing.visibility = "public"
        existing.team_id = None
        existing.custom_name = "owned"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"

        db = MagicMock()
        db.rollback = Mock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", return_value=existing),
            patch("mcpgateway.services.permission_service.PermissionService.check_resource_ownership", new=AsyncMock(return_value=False)),
        ):
            upd = PromptUpdate(description="new desc")

            with pytest.raises(PermissionError, match="Only the owner"):
                await prompt_service.update_prompt(db, 1, upd, user_email="other@test.com")

    @pytest.mark.asyncio
    async def test_update_integrity_error(self, prompt_service):
        """IntegrityError during update is propagated."""
        existing = _build_db_prompt(name="my-prompt")
        existing.visibility = "public"
        existing.team_id = None
        existing.custom_name = "my-prompt"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"

        db = MagicMock()
        db.commit = Mock(side_effect=IntegrityError("dup", None, BaseException()))
        db.rollback = Mock()

        with patch("mcpgateway.services.prompt_service.get_for_update", return_value=existing):
            upd = PromptUpdate(description="new desc")

            with pytest.raises(IntegrityError):
                await prompt_service.update_prompt(db, 1, upd)

    @pytest.mark.asyncio
    async def test_update_name_on_gateway_prompt(self, prompt_service):
        """Updating name on a gateway prompt sets custom_name instead of original_name."""
        existing = _build_db_prompt(name="gw__old")
        existing.visibility = "public"
        existing.team_id = None
        existing.custom_name = "old"
        existing.gateway = MagicMock()
        existing.gateway.name = "gw"
        existing.gateway_id = "gw-1"
        existing.owner_email = "owner@test.com"
        existing.version = 1

        db = MagicMock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", side_effect=[existing, None]),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            db.commit = Mock()
            db.refresh = Mock()
            prompt_service._notify_prompt_updated = AsyncMock()
            prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1})

            upd = PromptUpdate(name="new-name", custom_name="custom")

            await prompt_service.update_prompt(db, 1, upd)

        assert existing.custom_name == "custom"

    @pytest.mark.asyncio
    async def test_update_name_on_local_prompt_sets_original_name_and_custom_name_and_initializes_version(self, prompt_service):
        """Covers local prompt name handling (original_name/custom_name) and version initialization."""
        existing = _build_db_prompt(name="old-name")
        existing.visibility = "public"
        existing.team_id = None
        existing.custom_name = "old-name"
        existing.original_name = "old-name"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"
        existing.version = None  # force version init branch

        db = MagicMock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", side_effect=[existing, None]),  # prompt then conflict-check
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            db.commit = Mock()
            db.refresh = Mock()
            prompt_service._notify_prompt_updated = AsyncMock()
            prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1})

            upd = PromptUpdate(name="new-name")
            await prompt_service.update_prompt(db, 1, upd)

        assert existing.original_name == "new-name"
        assert existing.custom_name == "new-name"
        assert existing.version == 1

    @pytest.mark.asyncio
    async def test_update_prompt_ignores_client_owner_email_for_private_conflict_scope(self, prompt_service):
        """Private conflict scope must use persisted owner, not payload owner_email."""
        # Third-Party
        from sqlalchemy.dialects import sqlite

        existing = _build_db_prompt(name="old-name")
        existing.visibility = "private"
        existing.team_id = None
        existing.custom_name = "old-name"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"
        existing.version = 1

        captured_where = {}

        def fake_get_for_update(_db, _model, _prompt_id=None, **kwargs):  # noqa: ANN001, ANN202
            where_clause = kwargs.get("where")
            if where_clause is not None:
                captured_where["sql"] = str(where_clause.compile(dialect=sqlite.dialect(), compile_kwargs={"literal_binds": True}))
                return None
            return existing

        db = MagicMock()
        # Caller is an admin so the update is allowed; the test only verifies
        # that the persisted owner (not the payload owner_email) is used in
        # the WHERE clause.  Explicit is_user_admin patch avoids depending
        # on MagicMock-is-truthy in the real admin check.
        with (
            patch("mcpgateway.services.prompt_service.get_for_update", side_effect=fake_get_for_update),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
            patch("mcpgateway.utils.admin_check.is_user_admin", return_value=True),
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            db.commit = Mock()
            db.refresh = Mock()
            prompt_service._notify_prompt_updated = AsyncMock()
            prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1})

            upd = PromptUpdate(name="new-name", owner_email="attacker@test.com")
            await prompt_service.update_prompt(db, 1, upd, user_email="caller@test.com")

        assert "owner@test.com" in captured_where["sql"]
        assert "attacker@test.com" not in captured_where["sql"]

    @pytest.mark.asyncio
    async def test_update_arguments_without_description_skips_description_key(self, prompt_service):
        """Covers arg.description is None branch when rebuilding argument_schema."""
        existing = _build_db_prompt(name="my-prompt")
        existing.visibility = "public"
        existing.team_id = None
        existing.custom_name = "my-prompt"
        existing.gateway = None
        existing.gateway_id = None
        existing.owner_email = "owner@test.com"
        existing.version = 1

        db = MagicMock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", return_value=existing),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.cache.admin_stats_cache.admin_stats_cache") as mock_admin_cache,
        ):
            mock_cache = AsyncMock()
            mock_cache_fn.return_value = mock_cache
            mock_admin_cache.invalidate_tags = AsyncMock()
            db.commit = Mock()
            db.refresh = Mock()
            prompt_service._notify_prompt_updated = AsyncMock()
            prompt_service.convert_prompt_to_read = Mock(return_value={"id": 1})

            upd = PromptUpdate(arguments=[PromptArgument(name="user")])
            await prompt_service.update_prompt(db, 1, upd)

        assert existing.argument_schema["properties"]["user"] == {"type": "string"}


# ---------------------------------------------------------------------------
#  Additional coverage: set_prompt_state lock conflict + permission
# ---------------------------------------------------------------------------


class TestSetPromptStateLockAndPermission:
    """Cover lines 2044-2057: lock conflict and permission check in set_prompt_state."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_lock_conflict(self, prompt_service):
        """OperationalError during row lock raises PromptLockConflictError."""
        # Third-Party
        from sqlalchemy.exc import OperationalError

        # First-Party
        from mcpgateway.services.prompt_service import PromptLockConflictError

        db = MagicMock()
        db.rollback = Mock()

        with patch("mcpgateway.services.prompt_service.get_for_update", side_effect=OperationalError("locked", None, BaseException())):
            with pytest.raises(PromptLockConflictError, match="currently being modified"):
                await prompt_service.set_prompt_state(db, 1, activate=True)

    @pytest.mark.asyncio
    async def test_permission_denied(self, prompt_service):
        """Non-owner user gets PermissionError when toggling prompt state."""
        prompt = _build_db_prompt()
        prompt.enabled = True

        db = MagicMock()
        db.rollback = Mock()

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", return_value=prompt),
            patch("mcpgateway.services.permission_service.PermissionService.check_resource_ownership", new=AsyncMock(return_value=False)),
        ):
            with pytest.raises(PermissionError, match="Only the owner"):
                await prompt_service.set_prompt_state(db, 1, activate=False, user_email="other@test.com")


class TestSetPromptStateAdditionalBranches:
    """Covers skip_cache_invalidation and 'no-op' state updates."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_no_state_change_skips_commit_and_notifications(self, prompt_service):
        prompt = _build_db_prompt()
        prompt.enabled = True

        db = MagicMock()
        db.commit = Mock()
        db.refresh = Mock()

        prompt_service._notify_prompt_activated = AsyncMock()
        prompt_service._notify_prompt_deactivated = AsyncMock()
        prompt_service.convert_prompt_to_read = Mock(return_value={"enabled": True})
        prompt_service._get_team_name = Mock(return_value=None)

        with patch("mcpgateway.services.prompt_service.get_for_update", return_value=prompt):
            res = await prompt_service.set_prompt_state(db, 1, activate=True)

        db.commit.assert_not_called()
        prompt_service._notify_prompt_activated.assert_not_called()
        prompt_service._notify_prompt_deactivated.assert_not_called()
        assert res["enabled"] is True

    @pytest.mark.asyncio
    async def test_skip_cache_invalidation_true_does_not_touch_registry_cache(self, prompt_service):
        prompt = _build_db_prompt()
        prompt.enabled = True

        db = MagicMock()
        db.commit = Mock()
        db.refresh = Mock()

        prompt_service._notify_prompt_deactivated = AsyncMock()
        prompt_service.convert_prompt_to_read = Mock(return_value={"enabled": False})
        prompt_service._get_team_name = Mock(return_value=None)

        with (
            patch("mcpgateway.services.prompt_service.get_for_update", return_value=prompt),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
        ):
            res = await prompt_service.set_prompt_state(db, 1, activate=False, skip_cache_invalidation=True)

        mock_cache_fn.assert_not_called()
        prompt_service._notify_prompt_deactivated.assert_awaited_once()
        assert res["enabled"] is False


# ---------------------------------------------------------------------------
#  Additional coverage: event notification methods
# ---------------------------------------------------------------------------


class TestPromptEventNotifications:
    """Cover lines 2501-2586: _notify_prompt_* event publishing methods."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_notify_prompt_added(self, prompt_service):
        prompt = _build_db_prompt(pid=1, name="test")
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_added(prompt)

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_added"
        assert event["data"]["name"] == "test"

    @pytest.mark.asyncio
    async def test_notify_prompt_updated(self, prompt_service):
        prompt = _build_db_prompt(pid=2, name="updated")
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_updated(prompt)

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_updated"

    @pytest.mark.asyncio
    async def test_notify_prompt_activated(self, prompt_service):
        prompt = _build_db_prompt(pid=3, name="active")
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_activated(prompt)

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_activated"
        assert event["data"]["enabled"] is True

    @pytest.mark.asyncio
    async def test_notify_prompt_deactivated(self, prompt_service):
        prompt = _build_db_prompt(pid=4, name="inactive")
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_deactivated(prompt)

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_deactivated"
        assert event["data"]["enabled"] is False

    @pytest.mark.asyncio
    async def test_notify_prompt_deleted(self, prompt_service):
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_deleted({"id": 5, "name": "deleted"})

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_deleted"

    @pytest.mark.asyncio
    async def test_notify_prompt_removed(self, prompt_service):
        prompt = _build_db_prompt(pid=6, name="removed")
        prompt_service._event_service.publish_event = AsyncMock()

        await prompt_service._notify_prompt_removed(prompt)

        event = prompt_service._event_service.publish_event.call_args[0][0]
        assert event["type"] == "prompt_removed"
        assert event["data"]["enabled"] is False


class TestPromptSubscribeEvents:
    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_subscribe_events_empty_generator_exits(self, prompt_service):
        async def _empty_events():
            if False:
                yield {"type": "never"}

        prompt_service._event_service.subscribe_events = _empty_events

        events = []
        async for event in prompt_service.subscribe_events():
            events.append(event)

        assert events == []


# ---------------------------------------------------------------------------
#  Additional coverage: register_prompts_bulk chunk exception
# ---------------------------------------------------------------------------


class TestRegisterPromptsBulkChunkException:
    """Cover lines 938-943: chunk-level exception in register_prompts_bulk."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_chunk_commit_exception(self, prompt_service):
        """Exception during chunk commit is caught and recorded in stats."""
        db = MagicMock()
        # Empty existing prompts list
        db.execute.return_value.scalars.return_value.all.return_value = []
        db.add_all = MagicMock()
        db.commit = MagicMock(side_effect=RuntimeError("db crash"))
        db.rollback = MagicMock()
        prompt_service._notify_prompt_added = AsyncMock()

        prompt = PromptCreate(
            name="test-prompt",
            description="desc",
            template="Hello {{ name }}",
            arguments=[],
        )

        result = await prompt_service.register_prompts_bulk(
            db=db,
            prompts=[prompt],
            created_by="tester",
            conflict_strategy="skip",
        )

        assert result["failed"] >= 1
        assert any("Chunk processing failed" in err for err in result["errors"])


# --------------------------------------------------------------------------- #
#                   Gateway ID Filtering Tests (#3638)                        #
# --------------------------------------------------------------------------- #


class TestListPromptsGatewayIdFilter:
    """Tests for gateway_id filtering in list_prompts."""

    @pytest.fixture
    def prompt_service(self):
        return PromptService()

    @pytest.mark.asyncio
    async def test_list_prompts_gateway_id_filter(self, prompt_service):
        """gateway_id filter should add a WHERE clause matching the gateway ID."""
        db = MagicMock()
        db.commit = Mock()
        prompt_service.convert_prompt_to_read = Mock(return_value=MagicMock())

        captured_query = None

        async def capture_paginate(db, query, **kwargs):
            nonlocal captured_query
            captured_query = query
            return ([], None)

        with patch("mcpgateway.services.prompt_service.unified_paginate", side_effect=capture_paginate):
            result, next_cursor = await prompt_service.list_prompts(db, gateway_id="some-gateway-id")

        assert result == []
        assert next_cursor is None
        # Verify the compiled query contains the gateway_id equality filter
        compiled = str(captured_query.compile(compile_kwargs={"literal_binds": True}))
        assert "some-gateway-id" in compiled

    @pytest.mark.asyncio
    async def test_list_prompts_gateway_id_null_filter(self, prompt_service):
        """gateway_id='null' should add an IS NULL WHERE clause."""
        db = MagicMock()
        db.commit = Mock()
        prompt_service.convert_prompt_to_read = Mock(return_value=MagicMock())

        captured_query = None

        async def capture_paginate(db, query, **kwargs):
            nonlocal captured_query
            captured_query = query
            return ([], None)

        with patch("mcpgateway.services.prompt_service.unified_paginate", side_effect=capture_paginate):
            result, next_cursor = await prompt_service.list_prompts(db, gateway_id="null")

        assert result == []
        assert next_cursor is None
        compiled = str(captured_query.compile(compile_kwargs={"literal_binds": True}))
        assert "gateway_id IS NULL" in compiled

    @pytest.mark.asyncio
    async def test_list_prompts_gateway_id_null_case_insensitive(self, prompt_service):
        """gateway_id='NULL' (uppercase) should also add an IS NULL WHERE clause."""
        db = MagicMock()
        db.commit = Mock()
        prompt_service.convert_prompt_to_read = Mock(return_value=MagicMock())

        captured_query = None

        async def capture_paginate(db, query, **kwargs):
            nonlocal captured_query
            captured_query = query
            return ([], None)

        with patch("mcpgateway.services.prompt_service.unified_paginate", side_effect=capture_paginate):
            result, next_cursor = await prompt_service.list_prompts(db, gateway_id="NULL")

        assert result == []
        assert next_cursor is None
        compiled = str(captured_query.compile(compile_kwargs={"literal_binds": True}))
        assert "gateway_id IS NULL" in compiled

    @pytest.mark.asyncio
    async def test_list_prompts_gateway_id_nonexistent_returns_empty(self, prompt_service):
        """Nonexistent gateway_id should return empty list, not an error."""
        db = MagicMock()
        db.commit = Mock()
        prompt_service.convert_prompt_to_read = Mock(return_value=MagicMock())

        captured_query = None

        async def capture_paginate(db, query, **kwargs):
            nonlocal captured_query
            captured_query = query
            return ([], None)

        with patch("mcpgateway.services.prompt_service.unified_paginate", side_effect=capture_paginate):
            result, next_cursor = await prompt_service.list_prompts(db, gateway_id="nonexistent-id")

        assert result == []
        assert next_cursor is None
        compiled = str(captured_query.compile(compile_kwargs={"literal_binds": True}))
        assert "nonexistent-id" in compiled

    @pytest.mark.asyncio
    async def test_list_prompts_gateway_id_included_in_cache_hash(self, prompt_service):
        """gateway_id should be part of the cache hash to prevent cache poisoning."""
        db = MagicMock()
        db.commit = Mock()

        with patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn:
            mock_cache = AsyncMock()
            mock_cache.hash_filters = MagicMock(return_value="hash123")
            mock_cache.get = AsyncMock(return_value=None)
            mock_cache.set = AsyncMock()
            mock_cache_fn.return_value = mock_cache

            with patch("mcpgateway.services.prompt_service.unified_paginate", new=AsyncMock(return_value=([], None))):
                await prompt_service.list_prompts(db, gateway_id="gw-123")

            # Verify gateway_id was passed to hash_filters
            mock_cache.hash_filters.assert_called_once()
            call_kwargs = mock_cache.hash_filters.call_args[1]
            assert call_kwargs.get("gateway_id") == "gw-123"

    @pytest.mark.asyncio
    async def test_list_prompts_without_gateway_id_no_filter(self, prompt_service):
        """When gateway_id is None, no gateway filtering should be applied."""
        db = MagicMock()
        db.commit = Mock()
        mock_prompt = _build_db_prompt()
        mock_prompt.team_id = None

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache_fn.return_value = AsyncMock(hash_filters=MagicMock(return_value="h"), get=AsyncMock(return_value=None), set=AsyncMock())
            mock_paginate.return_value = ([mock_prompt], None)

            result, _ = await prompt_service.list_prompts(db, gateway_id=None)

        assert result == ["converted"]

    @pytest.mark.asyncio
    async def test_list_prompts_creates_span(self, prompt_service):
        db = MagicMock()
        db.commit = Mock()
        prompt = _build_db_prompt()
        prompt.team_id = None

        span_cm = MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock(return_value=False))

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service.create_span", return_value=span_cm) as mock_create_span,
            patch("mcpgateway.services.prompt_service._get_registry_cache") as mock_cache_fn,
            patch.object(prompt_service, "_apply_access_control", new=AsyncMock(side_effect=lambda query, *_args, **_kwargs: query)),
            patch("mcpgateway.services.prompt_service.unified_paginate", new_callable=AsyncMock) as mock_paginate,
        ):
            mock_cache_fn.return_value = AsyncMock(hash_filters=MagicMock(return_value="h"), get=AsyncMock(return_value=None), set=AsyncMock())
            mock_paginate.return_value = ([prompt], None)

            result, _ = await prompt_service.list_prompts(db, user_email="user@example.com", token_teams=["team-1"], visibility="team")

        assert result == ["converted"]
        assert mock_create_span.call_args[0][0] == "prompt.list"
        attrs = mock_create_span.call_args[0][1]
        assert attrs["user.email"] == "user@example.com"
        assert attrs["team.scope"] == "team-1"
        assert attrs["visibility"] == "team"

    @pytest.mark.asyncio
    async def test_list_server_prompts_creates_span(self, prompt_service):
        db = MagicMock()
        db.commit = Mock()
        db.execute.return_value.scalars.return_value.all.return_value = [MagicMock(team_id=None)]

        span_cm = MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock(return_value=False))

        with (
            patch.object(prompt_service, "convert_prompt_to_read", return_value="converted"),
            patch("mcpgateway.services.prompt_service.create_span", return_value=span_cm) as mock_create_span,
        ):
            result = await prompt_service.list_server_prompts(db, "server-1", user_email="user@example.com", token_teams=["team-1"])

        assert result == ["converted"]
        assert mock_create_span.call_args[0][0] == "prompt.list"
        attrs = mock_create_span.call_args[0][1]
        assert attrs["server_id"] == "server-1"
        assert attrs["team.scope"] == "team-1"


# --------------------------------------------------------------------------- #
# Security regression tests for meta_data handling (CWE-400, CWE-20, CWE-284)#
# --------------------------------------------------------------------------- #


class TestValidateMetaDataPrompt:
    """Unit tests for _validate_meta_data in prompt_service (CWE-400 guards)."""

    def test_none_is_accepted(self):
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        _validate_meta_data(None)

    def test_empty_dict_is_accepted(self):
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        _validate_meta_data({})

    def test_valid_small_dict_is_accepted(self):
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        _validate_meta_data({"trace_id": "abc", "user": "test@example.com"})

    def test_too_many_keys_raises(self):
        """meta_data with more than _META_MAX_KEYS keys must be rejected."""
        # First-Party
        from mcpgateway.common.validators import META_MAX_KEYS
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        oversized = {str(i): i for i in range(META_MAX_KEYS + 1)}
        with pytest.raises(ValueError, match="maximum key count"):
            _validate_meta_data(oversized)

    def test_excessive_nesting_depth_raises(self):
        """meta_data with depth > _META_MAX_DEPTH must be rejected."""
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        deeply_nested = {"level1": {"level2": {"level3": "value"}}}
        with pytest.raises(ValueError, match="maximum nesting depth"):
            _validate_meta_data(deeply_nested)

    def test_list_of_dicts_depth_bypass_is_rejected(self):
        """Depth check must traverse lists so {"k": [{"l2": {"l3": "x"}}]} is rejected (CWE-400 / Finding 4)."""
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        hidden_depth = {"k": [{"l2": {"l3": "x"}}]}
        with pytest.raises(ValueError, match="maximum nesting depth"):
            _validate_meta_data(hidden_depth)

    def test_non_serializable_value_raises(self):
        """meta_data with non-JSON-serializable value must raise ValueError (CWE-20 / Finding 6)."""
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        class _Unserializable:
            pass

        with pytest.raises(ValueError, match="not serializable"):
            _validate_meta_data({"bad": _Unserializable()})

    def test_exact_max_depth_is_accepted(self):
        # First-Party
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        two_levels = {"outer": {"inner": "value"}}
        _validate_meta_data(two_levels)

    def test_oversized_bytes_raises(self):
        # First-Party
        from mcpgateway.common.validators import META_MAX_BYTES
        from mcpgateway.common.validators import validate_meta_data as _validate_meta_data

        large_value = "x" * (META_MAX_BYTES + 1)
        with pytest.raises(ValueError, match="maximum size"):
            _validate_meta_data({"k": large_value})


class TestBuildGetPromptRequest:
    """Unit tests for _build_get_prompt_request (CWE-20 and CWE-284 guards)."""

    def test_meta_is_injected_under_alias_key(self):
        """_meta must be present on the inner params model after build (CWE-20)."""
        # First-Party
        from mcpgateway.services.prompt_service import _build_get_prompt_request

        meta_data = {"trace_id": "xyz", "user": "alice@example.com"}
        request = _build_get_prompt_request("my-prompt", None, meta_data)
        inner_params = request.root.params
        assert inner_params is not None
        assert inner_params.meta is not None
        dumped = inner_params.meta.model_dump()
        # All meta_data keys must survive; MCP SDK may add progressToken alongside
        assert meta_data.items() <= dumped.items()

    def test_returns_client_request_type(self):
        """Return value must be a ClientRequest wrapping GetPromptRequest."""
        # Third-Party
        from mcp import types
        from mcp.types import GetPromptRequest

        # First-Party
        from mcpgateway.services.prompt_service import _build_get_prompt_request

        req = _build_get_prompt_request("my-prompt", {"arg": "val"}, {"k": "v"})
        assert isinstance(req, types.ClientRequest)
        assert isinstance(req.root, GetPromptRequest)


class TestGetPromptMetaDataValidationIntegration:
    """Integration tests: _validate_meta_data is called by _fetch_gateway_prompt_result."""

    @pytest.mark.asyncio
    async def test_fetch_gateway_prompt_rejects_oversized_meta_data(self):
        """_fetch_gateway_prompt_result must raise ValueError for oversized meta_data."""
        # First-Party
        from mcpgateway.common.validators import META_MAX_KEYS as _META_MAX_KEYS
        from mcpgateway.services.prompt_service import PromptService

        service = PromptService()

        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.url = "http://gateway.example.com/mcp"
        gateway.transport = "streamable_http"
        gateway.auth_type = None

        prompt = _build_db_prompt(template="", name="gw-prompt")
        prompt.gateway_id = "gw-1"
        prompt.gateway = gateway
        prompt.original_name = "gw-prompt"

        oversized = {str(i): i for i in range(_META_MAX_KEYS + 1)}

        with pytest.raises(ValueError, match="maximum key count"):
            await service._fetch_gateway_prompt_result(prompt, {}, meta_data=oversized)


class TestFetchGatewayPromptRegistryPath:
    """Covers the upstream-session-registry branches in _fetch_gateway_prompt_result (#4205)."""

    def _build_gateway_prompt(self):
        gateway = MagicMock()
        gateway.id = "gw-1"
        gateway.url = "http://gateway.example.com/mcp"
        gateway.transport = "streamable_http"
        gateway.auth_type = None
        gateway.auth_query_params = None

        prompt = _build_db_prompt(template="", name="gw-prompt")
        prompt.gateway_id = "gw-1"
        prompt.gateway = gateway
        prompt.original_name = "gw-prompt"
        return prompt

    @pytest.mark.asyncio
    async def test_fetch_gateway_prompt_uses_registry_when_downstream_session_id_present(self):
        """Registry path runs when Mcp-Session-Id is in scope and the registry is initialised.

        Covers prompt_service.py:402-403, 405, 408-409, 416.
        """
        # First-Party
        from mcpgateway.services.prompt_service import PromptService

        service = PromptService()
        prompt = self._build_gateway_prompt()

        remote_result = MagicMock()
        remote_result.messages = []
        remote_result.description = "from registry"

        fake_upstream = MagicMock()
        fake_upstream.session = MagicMock()

        class _Ctx:
            async def __aenter__(self):
                return fake_upstream

            async def __aexit__(self, *_exc):
                return False

        fake_registry = MagicMock()
        fake_registry.acquire = MagicMock(return_value=_Ctx())

        with (
            patch("mcpgateway.services.prompt_service._downstream_session_id_from_request", return_value="downstream-xyz"),
            patch("mcpgateway.services.prompt_service.get_upstream_session_registry", return_value=fake_registry),
            patch("mcpgateway.services.prompt_service._get_prompt_with_meta", new_callable=AsyncMock, return_value=remote_result),
        ):
            result = await service._fetch_gateway_prompt_result(prompt, {"a": "b"}, meta_data=None)

        fake_registry.acquire.assert_called_once()
        assert result.description == "from registry"

    @pytest.mark.asyncio
    async def test_fetch_gateway_prompt_falls_back_when_registry_not_initialised(self):
        """RegistryNotInitializedError falls through to the transport-direct path (covers line 406-407)."""
        # First-Party
        from mcpgateway.services.prompt_service import PromptService
        from mcpgateway.services.upstream_session_registry import RegistryNotInitializedError

        service = PromptService()
        prompt = self._build_gateway_prompt()

        remote_result = MagicMock()
        remote_result.messages = []
        remote_result.description = "from fallback"

        class _FakeStreamsCtx:
            async def __aenter__(self):
                # streamablehttp_client yields (read, write, get_session_id)
                return (MagicMock(), MagicMock(), MagicMock())

            async def __aexit__(self, *_exc):
                return False

        class _FakeClientSessionCtx:
            async def __aenter__(self):
                session = MagicMock()
                session.initialize = AsyncMock()
                return session

            async def __aexit__(self, *_exc):
                return False

        with (
            patch("mcpgateway.services.prompt_service._downstream_session_id_from_request", return_value="downstream-xyz"),
            patch("mcpgateway.services.prompt_service.get_upstream_session_registry", side_effect=RegistryNotInitializedError("not init")),
            patch("mcpgateway.services.prompt_service.streamablehttp_client", return_value=_FakeStreamsCtx()),
            patch("mcpgateway.services.prompt_service.ClientSession", return_value=_FakeClientSessionCtx()),
            patch("mcpgateway.services.prompt_service._get_prompt_with_meta", new_callable=AsyncMock, return_value=remote_result),
        ):
            result = await service._fetch_gateway_prompt_result(prompt, None, meta_data=None)

        assert result.description == "from fallback"
