praxis/tests/test_p1_guardrail_e2e.py

"""P1 integration test — guardrail e2e through the assist pipeline (TASK-08-03).

Verifies REQ-ASSIST-03 (the guardrail works in the pipeline, not just standalone):
  1. Start a shift.
  2. Mock an LLM response that gives a direct answer → guardrail blocks it +
     canned fallback is sent to TTS.
  3. The turn's guardrail_verdict_json has allowed=False, category='blocked_direct_script'.
  4. guardrail_block_count is incremented.
  5. Mock an LLM response that gives a coaching question → allowed + sent to TTS.
  6. The turn's guardrail_verdict_json has allowed=True, category='coaching'.
  7. Incremental audit-log: the partial turn (ASR only) is written before the
     LLM response, then updated with the LLM response + verdict (REQ-IDEATE-09).
"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from db.migrate import apply_migrations
from db.store import PraxisStore, HARDCODED_LEARNER_ID
from server.assist.context import AssistContext, COACHING_INSTRUCTION
from server.assist.guardrail_processor import LiveAssistGuardrailProcessor
from server.assist.routes import router as assist_router
from server.assist.session import AssistSession
from server.guardrails.live_assist import CANNED_FALLBACK, LiveAssistGuardrail
from server.services.base import GuardrailContext


@pytest.fixture
def store(tmp_path: Path) -> PraxisStore:
    db = tmp_path / "test_p1_guardrail_e2e.db"
    apply_migrations(db)
    s = PraxisStore(db)
    asyncio.run(s.init())
    return s


def _ctx() -> AssistContext:
    return AssistContext(
        system_prompt=f"{COACHING_INSTRUCTION}\n\nWeek 1, damaged-product refund.\n\nBe brief.",
        current_week=1,
        scenario_tag="damaged-product refund",
        theta=0.0,
        coaching_focus="empathy",
        path_slug="customer_service",
    )


def test_guardrail_blocks_direct_answer_e2e(store: PraxisStore):
    """A direct-answer LLM response is blocked + canned fallback is emitted (TASK-08-03)."""
    session = AssistSession(store, HARDCODED_LEARNER_ID, _ctx())
    asyncio.run(session.start())

    # Simulate the in-loop guardrail processor on a direct-answer LLM response.
    proc = LiveAssistGuardrailProcessor(
        guardrail=LiveAssistGuardrail(), session=session, llm_context=None
    )
    proc.push_frame = AsyncMock()

    async def _run():
        from pipecat.frames.frames import LLMFullResponseEndFrame, TextFrame, TranscriptionFrame

        # ASR transcript (partial turn — REQ-IDEATE-09).
        await proc.process_frame(
            TranscriptionFrame(text="Customer wants a refund", user_id="u", timestamp=""),
            direction=1,
        )
        # LLM response: direct answer.
        await proc.process_frame(TextFrame(text="You should say sorry to the customer."), direction=1)
        await proc.process_frame(LLMFullResponseEndFrame(), direction=1)

    asyncio.run(_run())

    # The block count was incremented.
    assert session.guardrail_block_count == 1
    # The canned fallback was emitted (pushed as a TextFrame).
    pushed_texts = [
        call.args[0].text for call in proc.push_frame.await_args_list
        if hasattr(call.args[0], "text")
    ]
    assert CANNED_FALLBACK in pushed_texts
    # The turn's guardrail_verdict_json has allowed=False.
    turns = asyncio.run(store.get_turns(session.session_id))
    assert len(turns) == 1
    verdict = json.loads(turns[0].guardrail_verdict_json)
    assert verdict["allowed"] is False
    assert verdict["category"] == "blocked_direct_script"


def test_guardrail_allows_coaching_question_e2e(store: PraxisStore):
    """A coaching-question LLM response is allowed + sent to TTS (TASK-08-03)."""
    session = AssistSession(store, HARDCODED_LEARNER_ID, _ctx())
    asyncio.run(session.start())

    proc = LiveAssistGuardrailProcessor(
        guardrail=LiveAssistGuardrail(), session=session, llm_context=None
    )
    proc.push_frame = AsyncMock()

    async def _run():
        from pipecat.frames.frames import LLMFullResponseEndFrame, TextFrame, TranscriptionFrame

        await proc.process_frame(
            TranscriptionFrame(text="Customer is upset", user_id="u", timestamp=""),
            direction=1,
        )
        await proc.process_frame(TextFrame(text="What do you think the customer needs?"), direction=1)
        await proc.process_frame(LLMFullResponseEndFrame(), direction=1)

    asyncio.run(_run())

    # No block.
    assert session.guardrail_block_count == 0
    # The turn's guardrail_verdict_json has allowed=True, category='coaching'.
    turns = asyncio.run(store.get_turns(session.session_id))
    assert len(turns) == 1
    verdict = json.loads(turns[0].guardrail_verdict_json)
    assert verdict["allowed"] is True
    assert verdict["category"] == "coaching"


def test_incremental_audit_log_partial_then_complete(store: PraxisStore):
    """REQ-IDEATE-09: partial turn (ASR) written before LLM response, then updated with verdict."""
    session = AssistSession(store, HARDCODED_LEARNER_ID, _ctx())
    asyncio.run(session.start())

    proc = LiveAssistGuardrailProcessor(
        guardrail=LiveAssistGuardrail(), session=session, llm_context=None
    )
    proc.push_frame = AsyncMock()

    async def _run_partial_only():
        from pipecat.frames.frames import TranscriptionFrame

        # ASR arrives but the LLM never responds (simulated abrupt termination).
        await proc.process_frame(
            TranscriptionFrame(text="Customer is upset", user_id="u", timestamp=""),
            direction=1,
        )

    asyncio.run(_run_partial_only())
    # The partial turn (ASR only) is in the turns table with tts_text NULL.
    turns = asyncio.run(store.get_turns(session.session_id))
    assert len(turns) == 1
    assert turns[0].asr_text == "Customer is upset"
    assert turns[0].tts_text is None
    assert turns[0].guardrail_verdict_json is None

    # Now simulate the LLM response arriving (the turn is completed).
    async def _run_complete():
        from pipecat.frames.frames import LLMFullResponseEndFrame, TextFrame

        await proc.process_frame(TextFrame(text="How could you acknowledge their frustration?"), direction=1)
        await proc.process_frame(LLMFullResponseEndFrame(), direction=1)

    asyncio.run(_run_complete())
    turns = asyncio.run(store.get_turns(session.session_id))
    # The partial turn was updated (not a new row).
    assert len(turns) == 1
    assert turns[0].tts_text is not None
    assert turns[0].guardrail_verdict_json is not None
    verdict = json.loads(turns[0].guardrail_verdict_json)
    assert verdict["allowed"] is True
    assert verdict["category"] == "coaching"