vocero-client/vocero-api/tests/test_content_context.py

"""Linked content, explicit conversation selection, and authenticated greeting context."""

import asyncio
import json
import os
import threading
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, mcp
    from vocero_api.realtime.mcp_proxy import McpRelay
    from vocero_api.services import content_context, conversations, mcp_reasoning, mcp_tools

REPORT_ID = "report-11111111-2222-4333-8444-555555555555"
TITLE = "Alimentación de La Leona"
OPENING_INSTRUCTION = "For the opening greeting only"


def report_result(report_id=REPORT_ID, title=TITLE, has_podcast=True):
    return {
        "data": {
            "result": {
                "status": "found",
                "detail": {
                    "report": {"id": report_id, "title": title, "date": "2026-10-08", "has_podcast": has_podcast},
                    "summary": "Full details must not be copied to the startup prompt.",
                },
            }
        }
    }


def podcast_result(podcast_id="daily:episode-1", title="Resumen diario"):
    return {
        "data": {"podcast": {"id": podcast_id, "kind": podcast_id.split(":")[0], "title": title, "date": "2026-10-08"}}
    }


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


@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, "S2S_START_SESSION_MODE", "TALKFIRST")
    monkeypatch.setattr(settings, "S2S_PRIMARY_PROMPT", "")
    monkeypatch.setattr(settings, "OPENAI_API_KEY", "fixture-key")
    monkeypatch.setattr(
        backend_login,
        "login",
        AsyncMock(return_value=BackendLogin(sessionToken="private-alice", user={"id": "alice", "name": "Alice"})),
    )
    monkeypatch.setattr(
        mcp,
        "discover",
        AsyncMock(return_value=[McpToolDefinition(name=name, parameters={}) for name in sorted(mcp.ALLOWED_TOOLS)]),
    )
    execute = AsyncMock(return_value=report_result())
    monkeypatch.setattr(mcp, "execute", execute)
    conversations.store.clear()
    yield execute
    conversations.store.clear()


def bootstrap(client, **selectors):
    response = client.post("/api/conversations", json={"phoneNumber": "+56000000001", **selectors})
    assert response.status_code == 200, response.text
    item = conversations.store.get(client.cookies.get(conversations.COOKIE_NAME), response.json()["conversationId"])
    return item


def test_report_link_is_resolved_once_with_the_caller_token_and_only_metadata_is_retained(enabled):
    with TestClient(app) as client:
        item = bootstrap(client, report_id=REPORT_ID)
        enabled.assert_awaited_once_with(
            "private-alice", "get_report", {"request": {"report_id": REPORT_ID, "limit": 1}}
        )
        assert item.linked_content.report.title == TITLE
        assert item.current_content.podcast.id == f"report:{REPORT_ID}"
        assert item.linked_content == item.current_content
        prompt = mcp_tools.instructions("Base instructions", item)
        assert TITLE in prompt and REPORT_ID in prompt and OPENING_INSTRUCTION in prompt
        assert "Full details" not in prompt and "private-alice" not in prompt
        catalog = client.get("/api/mcp/tools", headers={conversations.CONVERSATION_HEADER: item.id}).json()["tools"]
        assert {tool["name"] for tool in catalog} == mcp.ALLOWED_TOOLS | {"select_content"}


def test_no_link_needs_no_lookup_and_existing_clients_keep_working(enabled):
    with TestClient(app) as client:
        item = bootstrap(client)
        enabled.assert_not_called()
        assert item.current_content.report is None and item.current_content.podcast is None
        assert not item.current_content.unavailable


@pytest.mark.parametrize(
    "result",
    [
        {"error": "Denied: never expose a private title"},
        {"data": {"result": {"status": "ambiguous", "choices": []}}},
        report_result(report_id="some-other-report"),
        {"data": {"result": {"status": "found", "detail": {"report": {"id": REPORT_ID}}}}},
    ],
)
def test_unavailable_or_invalid_links_allow_a_generic_conversation_without_leaking_details(enabled, result):
    enabled.return_value = result
    with TestClient(app) as client:
        item = bootstrap(client, report_id=REPORT_ID)
        assert item.current_content.unavailable
        assert item.current_content.report is None and item.current_content.podcast is None
        prompt = mcp_tools.instructions("", item)
        assert TITLE not in prompt and "private title" not in prompt
        assert OPENING_INSTRUCTION in prompt


@pytest.mark.parametrize("status", [401, 403, 502])
def test_link_transport_failures_preserve_authentication_boundary(enabled, status):
    enabled.side_effect = IntegrationError(status, "Upstream detail must stay private")
    with TestClient(app) as client:
        response = client.post("/api/conversations", json={"phoneNumber": "+56000000001", "report_id": REPORT_ID})
        if status == 401:
            assert response.status_code == 401
            assert not conversations.store._items
        else:
            assert response.status_code == 200
            item = conversations.store.get(
                client.cookies.get(conversations.COOKIE_NAME), response.json()["conversationId"]
            )
            assert item.current_content.unavailable
            assert "Upstream detail" not in mcp_tools.instructions("", item)


def test_link_lookup_timeout_does_not_block_the_whole_conversation(enabled, monkeypatch):
    cancelled = []

    async def slow_lookup(*args):
        try:
            await asyncio.sleep(10)
        finally:
            cancelled.append(True)

    enabled.side_effect = slow_lookup
    monkeypatch.setattr(content_context, "LINK_LOOKUP_TIMEOUT_SEC", 0.01)
    with TestClient(app) as client:
        assert bootstrap(client, report_id=REPORT_ID).current_content.unavailable
    assert cancelled == [True]


@pytest.mark.parametrize("episode_available", [True, False])
def test_explicit_daily_episode_wins_without_falling_back_to_the_report_podcast(enabled, episode_available):
    enabled.side_effect = [report_result(), podcast_result() if episode_available else {"error": "Unavailable"}]
    with TestClient(app) as client:
        item = bootstrap(client, report_id=REPORT_ID, episode_id="episode-1")
        assert item.linked_content.report.id == REPORT_ID
        assert (item.linked_content.podcast.id if item.linked_content.podcast else None) == (
            "daily:episode-1" if episode_available else None
        )
        assert item.linked_content.unavailable is not episode_available
        assert enabled.await_args_list[1].args == (
            "private-alice",
            "get_podcast",
            {"request": {"podcast_id": "daily:episode-1"}},
        )


def test_linked_content_is_isolated_between_tabs_and_cleared_when_closed(enabled):
    enabled.side_effect = [report_result(), report_result("report-b", "Mortalidad", False)]
    with TestClient(app) as client:
        first = bootstrap(client, report_id=REPORT_ID)
        second = bootstrap(client, report_id="report-b")
        assert second.current_content.podcast is None
        client.delete("/api/conversations/current", headers={conversations.CONVERSATION_HEADER: first.id})
        assert first.linked_content.report is None and first.current_content.podcast is None
        assert second.current_content.report.id == "report-b"


def test_search_and_comparison_leave_selection_unchanged_but_explicit_selection_replaces_it(enabled):
    async def scenario():
        linked = await content_context.from_link("private-alice", REPORT_ID, None)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"), linked)
        item.catalog = await content_context.catalog(item.backend_token)
        turn = mcp_tools.ToolTurn(item, item.generation)
        enabled.return_value = report_result("report-b", "Mortalidad", False)
        await turn.call("list_reports", {"request": {"start_date": "2026-10-01"}})
        assert enabled.await_args.args == ("private-alice", "list_reports", {"request": {"start_date": "2026-10-01"}})
        await turn.call("get_report", {"request": {"report_id": "report-b"}})
        assert item.current_content == linked
        await turn.call("select_content", {"request": {"kind": "report", "id": "report-b"}})
        assert item.current_content.report.id == "report-b" and item.current_content.podcast is None
        assert item.linked_content == linked
        enabled.return_value = {"error": "Denied"}
        assert "error" in await turn.call("select_content", {"request": {"kind": "report", "id": "private-report"}})
        assert item.current_content.report.id == "report-b"
        await turn.call("get_podcast", {})
        assert enabled.await_args.kwargs == {"podcast_id": None}

    asyncio.run(scenario())


def test_explicit_podcast_arguments_are_preserved_and_reading_does_not_change_the_default(enabled):
    async def scenario():
        linked = await content_context.from_link("private-alice", REPORT_ID, None)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"), linked)
        item.catalog = await content_context.catalog(item.backend_token)
        enabled.return_value = podcast_result()
        turn = mcp_tools.ToolTurn(item, item.generation)
        await turn.call("get_podcast", {"request": {"podcast_id": "daily:episode-1"}})
        assert enabled.await_args.args[2] == {"request": {"podcast_id": "daily:episode-1"}}
        assert enabled.await_args.kwargs == {"podcast_id": f"report:{REPORT_ID}"}
        assert item.current_content == linked
        await turn.call("select_content", {"request": {"kind": "podcast", "id": "daily:episode-1"}})
        assert item.current_content.podcast.id == "daily:episode-1" and item.current_content.report is None
        await turn.call("read_podcast_transcript", {"request": {"query": "oxígeno"}})
        assert enabled.await_args.kwargs == {"podcast_id": "daily:episode-1"}

    asyncio.run(scenario())


def test_cancelled_selection_cannot_replace_the_current_reference(enabled):
    async def scenario():
        linked = await content_context.from_link("private-alice", REPORT_ID, None)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"), linked)
        item.catalog = await content_context.catalog(item.backend_token)
        started, release = asyncio.Event(), asyncio.Event()

        async def slow_result(*args):
            started.set()
            try:
                await release.wait()
            except asyncio.CancelledError:
                await release.wait()
            return report_result("report-b", "Mortalidad")

        enabled.side_effect = slow_result
        turn = mcp_tools.ToolTurn(item, item.generation)
        item.pending = asyncio.create_task(
            turn.call("select_content", {"request": {"kind": "report", "id": "report-b"}})
        )
        task = item.pending
        await started.wait()
        item.interrupt()
        release.set()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert item.current_content == linked

    asyncio.run(scenario())


def test_implicit_podcast_reads_use_the_new_selection_within_the_same_turn(enabled):
    async def scenario():
        linked = await content_context.from_link("private-alice", REPORT_ID, None)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"), linked)
        item.catalog = await content_context.catalog(item.backend_token)
        turn = mcp_tools.ToolTurn(item, item.generation)
        await turn.call("get_podcast", {})
        assert enabled.await_args.kwargs == {"podcast_id": f"report:{REPORT_ID}"}
        enabled.return_value = podcast_result()
        await turn.call("select_content", {"request": {"kind": "podcast", "id": "daily:episode-1"}})
        await turn.call("get_podcast", {})
        assert enabled.await_args.kwargs == {"podcast_id": "daily:episode-1"}
        with pytest.raises(IntegrationError):
            await turn.call("get_podcast", {})
        enabled.reset_mock()
        fresh = mcp_tools.ToolTurn(item, item.generation)
        assert "error" in await fresh.call("select_content", {"request": {"kind": "report", "id": "   "}})
        enabled.assert_not_called()
        assert item.current_content.podcast.id == "daily:episode-1"

    asyncio.run(scenario())


def test_revoked_authentication_stops_remaining_link_lookups(enabled):
    enabled.side_effect = IntegrationError(401, "La sesión expiró.")
    with TestClient(app) as client:
        response = client.post(
            "/api/conversations",
            json={"phoneNumber": "+56000000001", "report_id": REPORT_ID, "episode_id": "episode-1"},
        )
        assert response.status_code == 401
        assert enabled.await_count == 1
        assert not conversations.store._items


@pytest.mark.parametrize("mode", ["TALKFIRST", "TALK", ""])
def test_http_greeting_and_followup_use_server_context_and_refresh_after_selection(enabled, monkeypatch, mode):
    monkeypatch.setattr(settings, "S2S_START_SESSION_MODE", mode)
    model = AsyncMock(
        side_effect=[
            "Hola, Alice.",
            marker("select_content", {"request": {"kind": "report", "id": "report-b"}}),
            "Revisemos mortalidad.",
        ]
    )
    monkeypatch.setattr(mcp_reasoning, "response_text", model)
    with TestClient(app) as client:
        item = bootstrap(client, report_id=REPORT_ID)
        headers = {conversations.CONVERSATION_HEADER: item.id, "X-Vocero-Turn": "1"}
        assert client.post("/api/reasoning", headers=headers, json={"transcript": "hola vocero"}).status_code == 200
        opening = model.await_args_list[0].args[0][0]["content"]
        assert TITLE in opening
        assert (OPENING_INSTRUCTION in opening) is (mode == "TALKFIRST")
        assert not item.greeting_pending
        enabled.return_value = report_result("report-b", "Mortalidad")
        headers["X-Vocero-Turn"] = "2"
        assert (
            client.post(
                "/api/reasoning",
                headers=headers,
                json={"transcript": "Ahora revisemos el otro reporte", "report_id": "browser-spoof"},
            ).status_code
            == 200
        )
        current_prompt = model.await_args.args[0][0]["content"]
        assert '"id": "report-b"' in current_prompt and TITLE in current_prompt
        assert "browser-spoof" not in current_prompt and OPENING_INSTRUCTION not in current_prompt
        assert item.linked_content.report.id == REPORT_ID


def test_voice_selection_refreshes_instructions_before_the_next_response_without_new_speech_protocol(enabled):
    async def scenario():
        linked = await content_context.from_link("private-alice", REPORT_ID, None)
        item = conversations.store.create("browser", await backend_login.login("+56000000001"), linked)
        item.catalog = await content_context.catalog(item.backend_token)
        item.greeting_pending = False
        browser, speech = AsyncMock(), AsyncMock()
        relay = McpRelay(browser, speech, item)
        relay.base_instructions = "Existing voice instructions"
        enabled.return_value = report_result("report-b", "Mortalidad")
        await relay.tool_call("select_content", {"request": {"kind": "report", "id": "report-b"}}, relay.turn)
        events = [json.loads(call.args[0]) for call in speech.send.await_args_list]
        assert [event["type"] for event in events] == ["session.update", "conversation.item.create", "response.create"]
        assert "Existing voice instructions" in events[0]["session"]["instructions"]
        assert '"id": "report-b"' in events[0]["session"]["instructions"]
        assert TITLE in events[0]["session"]["instructions"]
        assert "private-alice" not in json.dumps(events)

    asyncio.run(scenario())


@pytest.mark.parametrize("available", [True, False])
def test_voice_talkfirst_has_verified_link_context_before_greeting_without_a_session_ack(
    enabled, monkeypatch, available
):
    monkeypatch.setattr(settings, "S2S_REASONING_TYPE", "PROXY")
    enabled.return_value = report_result() if available else {"error": "Inaccessible title must not be disclosed"}
    received, errors = [], []
    refreshed = threading.Event()

    def voice(socket):
        try:
            socket.send(json.dumps({"type": "session.created"}))
            for raw in socket:
                event = json.loads(raw)
                received.append(event)
                if event["type"] == "session.update" and OPENING_INSTRUCTION not in event["session"]["instructions"]:
                    refreshed.set()
                if event["type"] == "response.create":
                    socket.send(json.dumps({"type": "response.created", "response": {"id": "opening"}}))
                    socket.send(
                        json.dumps(
                            {
                                "type": "response.done",
                                "response": {
                                    "id": "opening",
                                    "status": "completed",
                                    "output": [{"content": [{"text": "Hola, Alice."}]}],
                                },
                            }
                        )
                    )
        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:
            item = bootstrap(client, report_id=REPORT_ID)
            with client.websocket_connect(f"/v1/realtime?conversation_id={item.id}") as socket:
                assert socket.receive_json()["type"] == "session.created"
                socket.send_json({"type": "session.update", "session": {"instructions": "Existing instructions"}})
                socket.send_json(
                    {
                        "type": "conversation.item.create",
                        "item": {
                            "role": "user",
                            "type": "message",
                            "content": [{"type": "input_text", "text": "hola vocero"}],
                        },
                    }
                )
                socket.send_json({"type": "response.create"})
                assert socket.receive_json()["type"] == "response.created"
                assert socket.receive_json()["type"] == "response.done"
                assert refreshed.wait(2)
                assert not item.greeting_pending
    thread.join(3)
    assert not errors
    assert [event["type"] for event in received[:3]] == [
        "session.update",
        "conversation.item.create",
        "response.create",
    ]
    initial = received[0]["session"]["instructions"]
    assert OPENING_INSTRUCTION in initial and "Existing instructions" in initial
    assert (TITLE in initial) is available
    assert "Inaccessible title" not in initial
    assert "private-alice" not in json.dumps(received)