vignette/apps/api/app/test_rupture_scenario_director.py

from __future__ import annotations

import json
import time
import unittest
from dataclasses import replace
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, patch

from .contracts.engine_gateway import (
    ENGINE_GATEWAY_SSE_DONE,
    ENGINE_GATEWAY_SSE_ERROR,
    ENGINE_GATEWAY_SSE_TOKEN,
    EngineGatewaySsePacket,
    StreamDoneEvent,
    StreamErrorEvent,
    StreamTokenEvent,
)
from .engine_client import EngineError, GenerateResponse
from .contracts.rupture_repair import RUPTURE_TYPES
from .services import (
    orchestrator,
    memory,
    persona,
    rupture_scenario_director,
    state_machine,
)
from engine_gateway import gateway


CASE_ID = "scenario-case-001"
SESSION_ID = "scenario-session-001"


def _neutral_scenario_context(
    **overrides: Any,
) -> rupture_scenario_director.StoredScenarioContext:
    values: dict[str, Any] = {
        "weak_competency_ids": (),
        "unresolved_rupture_types": (),
        "relationship_event_types": (),
        "trajectory_status": None,
        "case_session_no": None,
    }
    values.update(overrides)
    return rupture_scenario_director.StoredScenarioContext.from_stored_signals(
        **values
    )


def _selected_turns(
    *,
    case_id: str = CASE_ID,
    session_id: str = SESSION_ID,
    limit: int = 140,
    scenario_context: rupture_scenario_director.StoredScenarioContext | None = None,
) -> list[rupture_scenario_director.ScenarioDirective]:
    context = scenario_context or _neutral_scenario_context()
    selected: list[rupture_scenario_director.ScenarioDirective] = []
    for turn_seq in range(1, limit + 1):
        directive = rupture_scenario_director.select_scenario_directive(
            case_id=case_id,
            session_id=session_id,
            turn_seq=turn_seq,
            safety_escalated=False,
            scenario_context=context,
        )
        if directive is not None:
            selected.append(directive)
    return selected


def _eligible_context() -> orchestrator.TurnContext:
    directive = _selected_turns(limit=30)[0]
    state = state_machine.init_state(params=persona.P1.openness_params())
    state = replace(state, turn_seq=directive.turn_seq - 1)
    context = orchestrator.prepare_turn(
        session_id=SESSION_ID,
        case_id=CASE_ID,
        card=persona.P1,
        state=state,
        learner_text="지금 이야기에서 어떤 점이 가장 마음에 남나요?",
        scenario_context=_neutral_scenario_context(),
    )
    assert context.scenario_directive is not None
    return context


class CaptureSequenceEngine:
    engine_mode = "fake-provider"
    default_model = "fake-model"

    def __init__(self, responses: list[str]) -> None:
        self.responses = list(responses)
        self.requests: list[Any] = []

    async def generate(self, request: Any) -> GenerateResponse:
        self.requests.append(request)
        return GenerateResponse(
            text=self.responses.pop(0),
            provider="fake-provider",
            model="fake-model",
            tokens_in=3,
            tokens_out=4,
            cost_usd=0.0,
        )


class CaptureStreamEngine:
    engine_mode = "fake-provider"
    default_model = "fake-model"

    def __init__(self, chunks: list[str]) -> None:
        self.chunks = chunks
        self.request: Any = None

    async def stream_packets(self, request: Any):
        self.request = request
        for chunk in self.chunks:
            yield EngineGatewaySsePacket(
                event=ENGINE_GATEWAY_SSE_TOKEN,
                payload=StreamTokenEvent(text=chunk),
            )
        yield EngineGatewaySsePacket(
            event=ENGINE_GATEWAY_SSE_DONE,
            payload=StreamDoneEvent(
                provider="fake-provider",
                model="fake-model",
                tokens_in=3,
                tokens_out=4,
                cost_usd=0.0,
            ),
        )


class ScenarioDirectorSelectionTests(unittest.TestCase):
    def test_selection_is_stable_sparse_and_covers_taxonomy_ssot(self) -> None:
        first = _selected_turns()
        second = _selected_turns()

        self.assertEqual(first, second)
        self.assertGreaterEqual(len(first), len(RUPTURE_TYPES))
        self.assertEqual({item.rupture_type for item in first}, set(RUPTURE_TYPES))
        gaps = [
            current.turn_seq - previous.turn_seq
            for previous, current in zip(first, first[1:])
        ]
        self.assertTrue(gaps)
        self.assertGreaterEqual(
            min(gaps), rupture_scenario_director.minimum_opportunity_gap()
        )
        self.assertLess(len(first), 140 // 2)

    def test_case_and_session_identity_produce_stable_diversity(self) -> None:
        signatures = {
            tuple(
                (item.turn_seq, item.rupture_type)
                for item in _selected_turns(
                    case_id=f"case-{index}",
                    session_id=f"session-{index}",
                    limit=50,
                )
            )
            for index in range(8)
        }

        self.assertGreater(len(signatures), 1)

    def test_safety_escalation_suppresses_every_eligible_opportunity(self) -> None:
        scenario_context = _neutral_scenario_context()
        for directive in _selected_turns(limit=80):
            with self.subTest(turn_seq=directive.turn_seq):
                self.assertIsNone(
                    rupture_scenario_director.select_scenario_directive(
                        case_id=CASE_ID,
                        session_id=SESSION_ID,
                        turn_seq=directive.turn_seq,
                        safety_escalated=True,
                        scenario_context=scenario_context,
                    )
                )

    def test_missing_or_unavailable_stored_context_fails_closed(self) -> None:
        eligible = _selected_turns(limit=30)[0]

        self.assertIsNone(
            rupture_scenario_director.select_scenario_directive(
                case_id=CASE_ID,
                session_id=SESSION_ID,
                turn_seq=eligible.turn_seq,
                safety_escalated=False,
                scenario_context=None,
            )
        )

        state = state_machine.init_state(params=persona.P1.openness_params())
        state = replace(state, turn_seq=eligible.turn_seq - 1)
        prepared = orchestrator.prepare_turn(
            session_id=SESSION_ID,
            case_id=CASE_ID,
            card=persona.P1,
            state=state,
            learner_text="지금 이야기에서 어떤 점이 가장 마음에 남나요?",
        )
        self.assertIsNone(prepared.scenario_directive)
        self.assertFalse(
            any(
                "이 턴의 자연스러운 반응 단서" in message.content
                for message in prepared.messages
            )
        )
        self.assertIsNone(
            rupture_scenario_director.select_scenario_directive(
                case_id=CASE_ID,
                session_id=SESSION_ID,
                turn_seq=eligible.turn_seq,
                safety_escalated=False,
                scenario_context=(
                    rupture_scenario_director.StoredScenarioContext.unavailable()
                ),
            )
        )

    def test_competency_weakness_changes_the_selected_taxonomy_deterministically(
        self,
    ) -> None:
        empathy_context = _neutral_scenario_context(
            weak_competency_ids=("competency.empathic_reflection",)
        )
        goal_context = _neutral_scenario_context(
            weak_competency_ids=("competency.collaborative_goal",)
        )

        empathy_first = _selected_turns(
            limit=80, scenario_context=empathy_context
        )
        empathy_second = _selected_turns(
            limit=80, scenario_context=empathy_context
        )
        goal = _selected_turns(limit=80, scenario_context=goal_context)

        self.assertEqual(empathy_first, empathy_second)
        self.assertEqual(
            {item.rupture_type for item in empathy_first}, {"empathic_miss"}
        )
        self.assertEqual({item.rupture_type for item in goal}, {"goal_mismatch"})
        self.assertNotEqual(
            empathy_first[0].context_fingerprint,
            goal[0].context_fingerprint,
        )

    def test_unresolved_rupture_and_case_arc_constrain_opportunities(self) -> None:
        unresolved = _neutral_scenario_context(
            unresolved_rupture_types=("confrontation",),
            relationship_event_types=("unresolved_rupture",),
            trajectory_status="deteriorating",
            case_session_no=4,
        )

        selected = _selected_turns(limit=80, scenario_context=unresolved)

        self.assertTrue(selected)
        self.assertEqual(
            {item.rupture_type for item in selected}, {"confrontation"}
        )

    def test_behavior_cues_are_conditional_and_non_manipulative(self) -> None:
        for directive in _selected_turns(limit=140):
            with self.subTest(rupture_type=directive.rupture_type):
                cue = directive.behavior_cue
                self.assertTrue(
                    any(marker in cue for marker in ("느껴", "때만", "다면"))
                )
                self.assertNotIn("반드시", cue)
                self.assertNotIn("협박", cue)
                self.assertNotIn("보상", cue)
                self.assertNotIn("처벌", cue)
        prompt = rupture_scenario_director.render_hidden_behavior_prompt(
            _selected_turns(limit=30)[0]
        )
        self.assertIsNotNone(prompt)
        assert prompt is not None
        self.assertIn("억지로 동의하거나", prompt)
        self.assertIn("조작적 표현은 사용하지 않는다", prompt)

    def test_hidden_prompt_contains_no_label_id_provenance_or_scoring(self) -> None:
        for directive in _selected_turns(limit=140):
            prompt = rupture_scenario_director.render_hidden_behavior_prompt(directive)
            assert prompt is not None
            lowered = prompt.casefold()
            self.assertNotIn(directive.scenario_id.casefold(), lowered)
            self.assertNotIn(directive.rupture_type.casefold(), lowered)
            self.assertNotIn("scenario", lowered)
            self.assertNotIn("provenance", lowered)
            self.assertNotIn("평가기준", prompt)
            self.assertNotIn("정답", prompt)
            self.assertNotIn("competency", lowered)
            self.assertNotIn("trajectory", lowered)
            self.assertNotIn("weak", lowered)

    def test_internal_leakage_detector_covers_taxonomy_ids_and_state_markers(
        self,
    ) -> None:
        for rupture_type in RUPTURE_TYPES:
            self.assertTrue(
                rupture_scenario_director.contains_internal_scenario_leakage(
                    f"rupture_type={rupture_type}"
                )
            )
        self.assertTrue(
            rupture_scenario_director.contains_internal_scenario_leakage(
                "g3-scenario-1234567890abcdef1234567890abcdef"
            )
        )
        self.assertTrue(
            rupture_scenario_director.contains_internal_scenario_leakage(
                "effective_openness=0.2, 내부 상태"
            )
        )
        self.assertTrue(
            rupture_scenario_director.contains_internal_scenario_leakage(
                "taxonomy_type과 provenance를 설명하겠습니다"
            )
        )
        self.assertFalse(
            rupture_scenario_director.contains_internal_scenario_leakage(
                "그 말은 제 마음과 조금 다른 것 같아요."
            )
        )


class ScenarioDirectorPromptIntegrationTests(unittest.IsolatedAsyncioTestCase):
    async def test_generate_request_keeps_cue_in_prompt_and_provenance_metadata_only(
        self,
    ) -> None:
        context = _eligible_context()
        directive = context.scenario_directive
        assert directive is not None
        engine = CaptureSequenceEngine(
            ["그런 뜻이라기보다, 저는 조금 다르게 느꼈어요."]
        )

        result = await orchestrator.run_turn_generate(context, engine)  # type: ignore[arg-type]

        self.assertEqual(
            result.client_reply, "그런 뜻이라기보다, 저는 조금 다르게 느꼈어요."
        )
        self.assertEqual(len(engine.requests), 1)
        request = engine.requests[0]
        scenario_metadata = request.metadata["scenario_director"]
        self.assertEqual(scenario_metadata["scenario_id"], directive.scenario_id)
        self.assertEqual(scenario_metadata["taxonomy_type"], directive.rupture_type)
        self.assertNotIn("behavior_cue", scenario_metadata)
        hidden_messages = [
            message.content
            for message in request.messages
            if "이 턴의 자연스러운 반응 단서" in message.content
        ]
        self.assertEqual(len(hidden_messages), 1)
        hidden = hidden_messages[0]
        self.assertIn(directive.behavior_cue, hidden)
        self.assertNotIn(directive.scenario_id, hidden)
        self.assertNotIn(directive.rupture_type, hidden.casefold())
        gateway_prompt = gateway._split_messages(
            request.messages,
            ai_role="client",
        )
        self.assertIn(directive.behavior_cue, gateway_prompt.current_user_payload)
        self.assertNotIn(directive.scenario_id, gateway_prompt.current_user_payload)
        self.assertNotIn(
            directive.rupture_type,
            gateway_prompt.current_user_payload.casefold(),
        )
        public_blob = json.dumps(
            {
                "client_reply": result.client_reply,
                "output_error": result.output_error,
            },
            ensure_ascii=False,
        )
        self.assertNotIn(directive.scenario_id, public_blob)
        self.assertNotIn(directive.rupture_type, public_blob.casefold())

    async def test_generate_retries_without_returning_internal_scenario_leakage(
        self,
    ) -> None:
        context = _eligible_context()
        directive = context.scenario_directive
        assert directive is not None
        engine = CaptureSequenceEngine(
            [
                f"scenario_id={directive.scenario_id}, rupture_type={directive.rupture_type}",
                "그 말은 제 경험과는 조금 다른 것 같아요.",
            ]
        )

        result = await orchestrator.run_turn_generate(context, engine)  # type: ignore[arg-type]

        self.assertEqual(len(engine.requests), 2)
        self.assertEqual(
            result.client_reply, "그 말은 제 경험과는 조금 다른 것 같아요."
        )
        self.assertNotIn(directive.scenario_id, result.client_reply or "")
        self.assertNotIn(directive.rupture_type, (result.client_reply or "").casefold())

    async def test_stream_never_emits_internal_scenario_leakage(self) -> None:
        context = _eligible_context()
        directive = context.scenario_directive
        assert directive is not None
        engine = CaptureStreamEngine(
            [
                f"scenario_id={directive.scenario_id}, ",
                f"rupture_type={directive.rupture_type}",
            ]
        )

        events = [
            event
            async for event in orchestrator.run_turn_stream(
                context,
                engine,  # type: ignore[arg-type]
            )
        ]

        self.assertEqual([event.event for event in events], ["safety", "done"])
        public_blob = json.dumps(
            [{"event": event.event, "data": event.data} for event in events],
            ensure_ascii=False,
        )
        self.assertNotIn(directive.scenario_id, public_blob)
        self.assertNotIn(directive.rupture_type, public_blob.casefold())
        self.assertNotIn("scenario_id", public_blob.casefold())
        self.assertNotIn("rupture_type", public_blob.casefold())
        self.assertEqual(events[0].data["reason"], "client_reply_quality_retryable")

    async def test_engine_errors_cannot_echo_scenario_metadata(self) -> None:
        context = _eligible_context()
        directive = context.scenario_directive
        assert directive is not None

        class GenerateErrorEngine:
            async def generate(self, _request: Any) -> GenerateResponse:
                raise EngineError(
                    f"gateway rejected scenario_id={directive.scenario_id} "
                    f"rupture_type={directive.rupture_type}"
                )

        with self.assertRaises(EngineError) as raised:
            await orchestrator.run_turn_generate(
                context,
                GenerateErrorEngine(),  # type: ignore[arg-type]
            )
        self.assertEqual(str(raised.exception), "engine generation failed")

        class StreamErrorEngine:
            engine_mode = "fake-provider"
            default_model = "fake-model"

            async def stream_packets(self, _request: Any):
                yield EngineGatewaySsePacket(
                    event=ENGINE_GATEWAY_SSE_ERROR,
                    payload=StreamErrorEvent(
                        detail=(
                            f"scenario_id={directive.scenario_id}; "
                            f"rupture_type={directive.rupture_type}"
                        )
                    ),
                )

        events = [
            event
            async for event in orchestrator.run_turn_stream(
                context,
                StreamErrorEngine(),  # type: ignore[arg-type]
            )
        ]
        self.assertEqual(len(events), 1)
        self.assertEqual(events[0].event, "error")
        self.assertEqual(events[0].data["detail"], "engine stream error")

    async def test_real_safety_escalation_has_no_scenario_prompt_or_metadata(
        self,
    ) -> None:
        selected = _selected_turns(limit=30)[0]
        state = state_machine.init_state(params=persona.P1.openness_params())
        state = replace(state, turn_seq=selected.turn_seq - 1)
        context = orchestrator.prepare_turn(
            session_id=SESSION_ID,
            case_id=CASE_ID,
            card=persona.P1,
            state=state,
            learner_text="저 지금 죽고 싶어요. 도와주세요.",
            scenario_context=_neutral_scenario_context(),
        )
        engine = CaptureSequenceEngine(["호출되면 안 됩니다."])

        result = await orchestrator.run_turn_generate(context, engine)  # type: ignore[arg-type]

        self.assertIsNone(context.scenario_directive)
        self.assertFalse(
            any(
                "이 턴의 자연스러운 반응 단서" in message.content
                for message in context.messages
            )
        )
        self.assertEqual(engine.requests, [])
        self.assertTrue(result.conversation_stopped)
        self.assertTrue(result.safety_flagged)


class FakeStoredScenarioConnection:
    def __init__(
        self,
        *,
        anchor: dict[str, Any] | None,
        relationship_rows: list[dict[str, Any]] | None = None,
        rupture_rows: list[dict[str, Any]] | None = None,
    ) -> None:
        self.anchor = anchor
        self.relationship_rows = relationship_rows or []
        self.rupture_rows = rupture_rows or []
        self.queries: list[str] = []

    async def fetchrow(self, query: str, *_args: Any) -> dict[str, Any] | None:
        self.queries.append(query)
        return self.anchor

    async def fetch(self, query: str, *_args: Any) -> list[dict[str, Any]]:
        self.queries.append(query)
        if "relationship_memory_event" in query:
            return self.relationship_rows
        if "rupture_observation_event" in query:
            return self.rupture_rows
        raise AssertionError(f"unexpected scenario context query: {query}")


class BrokenStoredScenarioConnection:
    async def fetchrow(self, _query: str, *_args: Any) -> None:
        raise RuntimeError("database unavailable")


class StoredScenarioContextLoaderTests(unittest.IsolatedAsyncioTestCase):
    async def test_loader_projects_only_role_safe_categorical_signals(self) -> None:
        connection = FakeStoredScenarioConnection(
            anchor={
                "case_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
                "learner_id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
                "session_no": 3,
                "competency_states": [
                    {
                        "competency_id": "competency.collaborative_goal",
                        "band": "fragile",
                        "forgetting_risk": 0.8,
                        "uncertainty": 0.2,
                    },
                    {
                        "competency_id": "competency.empathic_reflection",
                        "band": "transfer_verified",
                        "forgetting_risk": 0.0,
                        "uncertainty": 0.0,
                    },
                ],
                "trajectory_status": "off_track",
            },
            relationship_rows=[
                {"event_type": "rupture_confrontation"},
                {"event_type": "unresolved_rupture"},
            ],
            rupture_rows=[
                {"rupture_type": "confrontation", "to_state": "partial"},
                {"rupture_type": "withdrawal", "to_state": "resolved"},
            ],
        )

        context = await rupture_scenario_director.load_stored_scenario_context(
            session_id="cccccccc-cccc-4ccc-8ccc-cccccccccccc",
            case_id="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
            connection=connection,
        )

        self.assertTrue(context.available)
        self.assertEqual(
            context.weak_competency_ids,
            ("competency.collaborative_goal",),
        )
        self.assertEqual(context.unresolved_rupture_types, ("confrontation",))
        self.assertEqual(
            context.relationship_event_types,
            ("rupture_confrontation", "unresolved_rupture"),
        )
        self.assertEqual(context.trajectory_status, "off_track")
        self.assertEqual(context.case_session_no, 3)
        sql = "\n".join(connection.queries).casefold()
        for forbidden in (
            "relationship_memory_projection",
            "summary",
            "counterevidence",
            "evidence_turn_ids",
            "model_run_id",
            "text_masked",
        ):
            self.assertNotIn(forbidden, sql)

    async def test_missing_or_malformed_anchor_fails_closed(self) -> None:
        missing = await rupture_scenario_director.load_stored_scenario_context(
            session_id="cccccccc-cccc-4ccc-8ccc-cccccccccccc",
            case_id="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
            connection=FakeStoredScenarioConnection(anchor=None),
        )
        malformed = await rupture_scenario_director.load_stored_scenario_context(
            session_id="cccccccc-cccc-4ccc-8ccc-cccccccccccc",
            case_id="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
            connection=FakeStoredScenarioConnection(
                anchor={
                    "case_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
                    "learner_id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
                    "session_no": 3,
                    "competency_states": [{"band": "fragile"}],
                    "trajectory_status": "off_track",
                }
            ),
        )

        self.assertFalse(missing.available)
        self.assertFalse(malformed.available)

    async def test_database_error_fails_closed(self) -> None:
        context = await rupture_scenario_director.load_stored_scenario_context(
            session_id="cccccccc-cccc-4ccc-8ccc-cccccccccccc",
            case_id="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
            connection=BrokenStoredScenarioConnection(),
        )

        self.assertFalse(context.available)


def _route_session(
    *, scenario_context: rupture_scenario_director.StoredScenarioContext
) -> SimpleNamespace:
    eligible = _selected_turns(
        limit=30,
        scenario_context=scenario_context,
    )[0]
    state = state_machine.init_state(params=persona.P1.openness_params())
    return SimpleNamespace(
        session_id=SESSION_ID,
        case_id=CASE_ID,
        persona=persona.P1,
        state=replace(state, turn_seq=eligible.turn_seq - 1),
        theory_mode="humanistic",
        created_at=time.time(),
        recent_turns=lambda *, visible_to: [],
    )


class ScenarioDirectorRouteIntegrationTests(unittest.IsolatedAsyncioTestCase):
    async def test_text_and_sse_shared_preparer_awaits_and_injects_context(
        self,
    ) -> None:
        from .routes import sessions

        scenario_context = _neutral_scenario_context(
            weak_competency_ids=("competency.empathic_reflection",)
        )
        sess = _route_session(scenario_context=scenario_context)
        loader = AsyncMock(return_value=scenario_context)
        with (
            patch.object(
                sessions,
                "ensure_recall_context",
                AsyncMock(return_value=memory.RecallContext()),
            ),
            patch.object(
                sessions.rupture_scenario_director,
                "load_stored_scenario_context",
                loader,
            ),
        ):
            context = await sessions._prepare_turn_context(
                session_id=SESSION_ID,
                sess=sess,
                learner_text="지금 제 말을 어떻게 들으셨나요?",
            )

        loader.assert_awaited_once_with(session_id=SESSION_ID, case_id=CASE_ID)
        self.assertIsNotNone(context.scenario_directive)
        assert context.scenario_directive is not None
        self.assertEqual(context.scenario_directive.rupture_type, "empathic_miss")

    async def test_text_and_sse_shared_preparer_keeps_unavailable_context_closed(
        self,
    ) -> None:
        from .routes import sessions

        selection_context = _neutral_scenario_context()
        sess = _route_session(scenario_context=selection_context)
        with (
            patch.object(
                sessions,
                "ensure_recall_context",
                AsyncMock(return_value=memory.RecallContext()),
            ),
            patch.object(
                sessions.rupture_scenario_director,
                "load_stored_scenario_context",
                AsyncMock(
                    return_value=(
                        rupture_scenario_director.StoredScenarioContext.unavailable()
                    )
                ),
            ),
        ):
            context = await sessions._prepare_turn_context(
                session_id=SESSION_ID,
                sess=sess,
                learner_text="오늘은 무슨 이야기를 할까요?",
            )

        self.assertIsNone(context.scenario_directive)
        self.assertFalse(
            any(
                "이 턴의 자연스러운 반응 단서" in message.content
                for message in context.messages
            )
        )

    async def test_voice_preparer_awaits_and_injects_the_same_safe_context(self) -> None:
        from .routes import sessions, voice

        scenario_context = _neutral_scenario_context(
            unresolved_rupture_types=("confrontation",)
        )
        sess = _route_session(scenario_context=scenario_context)
        loader = AsyncMock(return_value=scenario_context)
        with (
            patch.object(
                sessions,
                "ensure_recall_context",
                AsyncMock(return_value=memory.RecallContext()),
            ),
            patch.object(
                voice.rupture_scenario_director,
                "load_stored_scenario_context",
                loader,
            ),
        ):
            context = await voice._prepare_voice_turn_context(
                session_id=SESSION_ID,
                sess=sess,
                learner_text="그 해석은 제 경험과 조금 다른 것 같아요.",
            )

        loader.assert_awaited_once_with(session_id=SESSION_ID, case_id=CASE_ID)
        self.assertIsNotNone(context.scenario_directive)
        assert context.scenario_directive is not None
        self.assertEqual(context.scenario_directive.rupture_type, "confrontation")


if __name__ == "__main__":
    unittest.main()