"""Caller isolation, bounded tool loops, and the unchanged speech protocol."""

import ast
import asyncio
import json
import os
import re
import threading
from pathlib import Path
from unittest.mock import AsyncMock, patch

import pytest
from fastapi.testclient import TestClient
from websockets.sync.server import serve

with patch.dict(os.environ, {}, clear=True), patch("dotenv.load_dotenv", return_value=False):
    from vocero_api.core import config as settings
    from vocero_api.main import app
    from vocero_api.models.mcp import BackendLogin, IntegrationError, McpToolDefinition
    from vocero_api.providers import backend_login, caller, mcp
    from vocero_api.services import conversations, mcp_reasoning, mcp_tools


def catalog():
    return [
        McpToolDefinition(
            name=name,
            description=f"Published {name} content",
            parameters={
                "type": "object",
                "properties": {
                    "request": {
                        "type": "object",
                        "properties": {"report_id": {"type": "string"}, "offset": {"type": "integer"}},
                        "additionalProperties": False,
                    }
                },
                "additionalProperties": False,
            },
        )
        for name in ("list_reports", "get_report")
    ]


def marker(name, arguments):
    return f"<<<VOCERO_MCP_CALL:NAME={name},ARGUMENTS={json.dumps(arguments)}>>>"


def test_mcp_marker_survives_the_existing_speech_sanitizer_without_model_dependencies():
    # Execute only the pure sanitizer and its constants, not the GPU pipeline imports.
    path = Path(__file__).resolve().parents[2] / "speech_to_speech/LLM/lm_output_processor.py"
    tree = ast.parse(path.read_text())
    selected = [
        node
        for node in tree.body
        if (
            isinstance(node, ast.Assign)
            and isinstance(node.targets[0], ast.Name)
            and node.targets[0].id.startswith("_APIPILOT_")
        )
        or (isinstance(node, ast.FunctionDef) and node.name == "_sanitize_apipilot_before_client_and_tts")
    ]
    namespace = {"re": re}
    exec(compile(ast.Module(body=selected, type_ignores=[]), str(path), "exec"), namespace)
    sanitize = namespace["_sanitize_apipilot_before_client_and_tts"]
    selection = marker("list_reports", {})
    assert sanitize(selection) == (selection, False, "")
    assert sanitize(selection.replace("VOCERO_MCP", "APIPILOT"))[1] is True


@pytest.mark.parametrize(
    "text",
    ["<<<VOC", "<<<VOCERO_MCP_CALL:NAME=get_report,ARGUMENTS=not-json>>>", "Before " + marker("list_reports", {})],
)
def test_malformed_markers_are_not_treated_as_user_facing_answers(text):
    with pytest.raises(IntegrationError):
        mcp_tools.selection(text)


@pytest.fixture
def enabled(monkeypatch):
    monkeypatch.setattr(settings, "S2S_MCP_URL", "http://mcp.test/mcp")
    monkeypatch.setattr(settings, "S2S_REASONING_TYPE", "OPENAI")
    monkeypatch.setattr(settings, "OPENAI_API_KEY", "local-model-key")
    monkeypatch.setattr(settings, "S2S_PRIMARY_PROMPT", "")
    login = AsyncMock(
        return_value=BackendLogin.model_validate(
            {"sessionToken": "demo-private-alice", "user": {"id": "alice", "name": "Alice"}}
        )
    )
    monkeypatch.setattr(backend_login, "login", login)
    monkeypatch.setattr(mcp, "discover", AsyncMock(return_value=catalog()))
    monkeypatch.setattr(caller, "_login_by_phone", AsyncMock(side_effect=AssertionError("Legacy login")))
    monkeypatch.setattr(caller, "_fetch_podcast_episode", AsyncMock(side_effect=AssertionError("Unscoped episode")))
    conversations.store.clear()
    yield login
    conversations.store.clear()


def bootstrap(client):
    response = client.post("/api/conversations", json={"phoneNumber": "+56000000001"})
    assert response.status_code == 200
    return response.json()["conversationId"]


def test_http_chains_tools_retains_history_and_ignores_browser_identity(enabled, monkeypatch):
    model = AsyncMock(
        side_effect=[
            marker("list_reports", {}),
            marker("get_report", {"request": {"report_id": "r1"}}),
            "Resumen r1",
            "Seguimiento r1",
        ]
    )
    execute = AsyncMock(return_value={"data": {"result": {"id": "r1", "summary": "Accessible finding"}}})
    monkeypatch.setattr(mcp_reasoning, "response_text", model)
    monkeypatch.setattr(mcp, "execute", execute)
    with TestClient(app) as client:
        identity = bootstrap(client)
        headers = {conversations.CONVERSATION_HEADER: identity, "x-vocero-turn": "1"}
        body = {
            "transcript": "Resume el último reporte",
            "phoneNumber": "another-user",
            "episode_id": "private-episode",
        }
        assert client.post("/api/reasoning", headers=headers, json=body).json() == {"answer": "Resumen r1"}
        assert execute.await_args_list[0].args == ("demo-private-alice", "list_reports", {})
        assert execute.await_args_list[1].args == ("demo-private-alice", "get_report", {"request": {"report_id": "r1"}})
        assert client.post("/api/reasoning", headers=headers, json=body).status_code == 409
        headers["x-vocero-turn"] = "2"
        assert (
            client.post("/api/reasoning", headers=headers, json={"transcript": "¿Y la recomendación?"}).status_code
            == 200
        )
        messages = model.await_args_list[-1].args[0]
        assert any(message["content"] == "Resumen r1" for message in messages)
        prompt = messages[0]["content"]
        assert '"additionalProperties": false' in prompt
        assert '"request"' in prompt
        assert "Alice" in prompt
        assert "demo-private" not in json.dumps(messages)
        assert "private-episode" not in json.dumps(messages)
        assert client.post("/api/reasoning", json=body).status_code == 401
        assert client.get("/api/user/phone?phoneNumber=123").status_code == 409
        assert client.get("/api/user/reports").status_code == 409
        assert client.get("/api/apipilot/tools").status_code == 404
        assert client.post("/api/apipilot/execute", json={"name": "get_report"}).status_code == 404
        enabled.assert_awaited_once_with("+56000000001")


@pytest.mark.parametrize("status", [401, 403, 502])
def test_http_errors_fail_closed_and_revocation_removes_the_conversation(enabled, monkeypatch, status):
    monkeypatch.setattr(mcp, "discover", AsyncMock(side_effect=IntegrationError(status, "Consulta no disponible.")))
    model = AsyncMock()
    monkeypatch.setattr(mcp_reasoning, "response_text", model)
    with TestClient(app) as client:
        identity = bootstrap(client)
        response = client.post(
            "/api/reasoning",
            headers={conversations.CONVERSATION_HEADER: identity, "x-vocero-turn": "1"},
            json={"transcript": "Hola"},
        )
        assert response.status_code == status
        model.assert_not_called()
        if status == 401:
            with pytest.raises(IntegrationError):
                conversations.store.get(client.cookies.get(conversations.COOKIE_NAME), identity)


def test_tool_arguments_duplicates_and_budgets_are_checked_before_execution(enabled, monkeypatch):
    execute = AsyncMock(return_value={"data": {"items": [], "next_offset": None}})
    monkeypatch.setattr(mcp, "execute", execute)
    monkeypatch.setattr(settings, "S2S_MCP_MAX_CALLS_PER_TURN", 3)

    async def scenario():
        login = await backend_login.login("+56000000001")
        item = conversations.store.create("browser", login)
        item.catalog = catalog()
        turn = mcp_tools.ToolTurn(item, item.generation)
        assert "error" in await turn.call("delete_user", {})
        assert "error" in await turn.call("get_report", {"request": {"user_id": "bob"}})
        execute.assert_not_called()
        await turn.call("list_reports", {"request": {"offset": 0}})
        with pytest.raises(IntegrationError):
            await turn.call("list_reports", {"request": {"offset": 10}})
        fresh = mcp_tools.ToolTurn(item, item.interrupt())
        await fresh.call("list_reports", {})
        with pytest.raises(IntegrationError):
            await fresh.call("list_reports", {})
        assert execute.await_count == 2

    asyncio.run(scenario())


def test_cancelled_tool_result_cannot_reach_a_new_turn_even_if_transport_returns_late(enabled, monkeypatch):
    async def scenario():
        started = asyncio.Event()
        cancelled = asyncio.Event()
        release = asyncio.Event()

        async def execute(*args):
            started.set()
            try:
                await release.wait()
            except asyncio.CancelledError:
                cancelled.set()
                await release.wait()
            return {"data": "stale result"}

        monkeypatch.setattr(mcp, "execute", execute)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"))
        item.catalog = catalog()
        turn = mcp_tools.ToolTurn(item, item.generation)
        item.pending = asyncio.create_task(turn.call("get_report", {}))
        pending = item.pending
        await started.wait()
        item.interrupt()
        await cancelled.wait()
        release.set()
        with pytest.raises(asyncio.CancelledError):
            await pending
        assert item.history == []

    asyncio.run(scenario())


def test_interrupt_for_an_old_http_turn_cannot_cancel_a_new_turn(enabled):
    with TestClient(app) as client:
        identity = bootstrap(client)
        item = conversations.store.get(client.cookies.get(conversations.COOKIE_NAME), identity)
        item.client_turn = 2
        headers = {conversations.CONVERSATION_HEADER: identity, "x-vocero-turn": "1"}
        assert client.post("/api/conversations/current/interrupt", headers=headers).status_code == 204
        assert item.generation == 0
        headers["x-vocero-turn"] = "2"
        assert client.post("/api/conversations/current/interrupt", headers=headers).status_code == 204
        assert item.generation == 1


@pytest.mark.parametrize("speech_order", [False, True])
def test_voice_relay_chains_tools_across_turns_hides_markers_and_preserves_audio(enabled, monkeypatch, speech_order):
    monkeypatch.setattr(settings, "S2S_REASONING_TYPE", "PROXY")
    monkeypatch.setattr(settings, "S2S_PRIMARY_PROMPT", "Deployment prompt")
    execute = AsyncMock(return_value={"data": {"id": "r1", "summary": "Accessible finding"}})
    monkeypatch.setattr(mcp, "execute", execute)
    received = []
    paths = []
    errors = []

    def voice(socket):
        try:
            paths.append(socket.request.path)
            socket.send(json.dumps({"type": "session.created"}))
            count = 0
            for raw in socket:
                if isinstance(raw, bytes):
                    socket.send(raw)
                    continue
                event = json.loads(raw)
                received.append(event)
                if event["type"] == "session.update":
                    socket.send(raw)
                if event["type"] != "response.create":
                    continue
                step = count % 3
                name = "list_reports" if step == 0 else "get_report"
                text = (
                    marker(name, {} if step == 0 else {"request": {"report_id": "r1"}})
                    if step < 2
                    else "Resultado publicado"
                )
                response_id = f"response-{count}"
                count += 1
                if speech_order:
                    socket.send(
                        json.dumps(
                            {
                                "type": "response.output_audio_transcript.done",
                                "response_id": response_id,
                                "transcript": text,
                            }
                        )
                    )
                socket.send(json.dumps({"type": "response.created", "response": {"id": response_id, "output": None}}))
                # Audio before text must wait until a normal answer is identified.
                socket.send(
                    json.dumps(
                        {
                            "type": "response.audio.delta",
                            "response_id": response_id,
                            "delta": "hidden-tool-audio" if step < 2 else "answer-audio",
                        }
                    )
                )
                if not speech_order:
                    for part in (text[:3], text[3:]):
                        socket.send(
                            json.dumps({"type": "response.text.delta", "response_id": response_id, "delta": part})
                        )
                done = {
                    "type": "response.done",
                    "response": {
                        "id": response_id,
                        "status": "completed",
                        "output": None if speech_order else [{"content": [{"text": text}]}],
                    },
                }
                socket.send(json.dumps(done))
                socket.send(json.dumps(done))
        except Exception as exc:
            errors.append(exc)

    with serve(voice, "127.0.0.1", 0) as upstream:
        thread = threading.Thread(target=upstream.serve_forever, daemon=True)
        thread.start()
        monkeypatch.setattr(
            settings, "S2S_PROXY_TARGET", f"ws://127.0.0.1:{upstream.socket.getsockname()[1]}/v1/realtime"
        )
        with TestClient(app) as client:
            identity = bootstrap(client)
            with client.websocket_connect(f"/v1/realtime?conversation_id={identity}") as socket:
                assert socket.receive_json() == {"type": "session.created"}
                socket.send_json(
                    {"type": "session.update", "session": {"instructions": "Deployment prompt\nSession RAG notes"}}
                )
                prompt = socket.receive_json()["session"]["instructions"]
                assert '"request"' in prompt and '"offset"' in prompt
                assert "Alice" in prompt and "demo-private" not in prompt
                assert "Session RAG notes" in prompt
                socket.send_bytes(b"\x00\x01")
                assert socket.receive_bytes() == b"\x00\x01"
                for _ in range(5):
                    socket.send_json(
                        {
                            "type": "conversation.item.create",
                            "item": {
                                "type": "message",
                                "role": "user",
                                "content": [{"type": "input_text", "text": "Resume el reporte"}],
                            },
                        }
                    )
                    socket.send_json({"type": "response.create"})
                    events = []
                    while True:
                        event = socket.receive_json()
                        events.append(event)
                        assert "hidden-tool-audio" not in json.dumps(event)
                        assert "VOCERO_MCP" not in json.dumps(event)
                        if event["type"] == "response.done" and any(e.get("delta") == "answer-audio" for e in events):
                            break
                    assert sum(event["type"] == "vocero.tool.started" for event in events) == 2
                    assert any(event.get("delta") == "answer-audio" for event in events)
            assert execute.await_count == 10
            assert all(call.args[0] == "demo-private-alice" for call in execute.await_args_list)
            assert identity not in paths[0]
            assert "demo-private" not in json.dumps(received)
    thread.join(3)
    assert not errors


def test_proxy_configuration_rejects_a_direct_speech_bypass(enabled, monkeypatch):
    monkeypatch.setattr(settings, "S2S_REASONING_TYPE", "PROXY")
    monkeypatch.setattr(settings, "SPEECH_TO_SPEECH_URL", "ws://speech.test/realtime")
    with TestClient(app) as client:
        assert client.get("/api/config").status_code == 503