"""학습자·교수자용 속마음 요약(§8.2·§9) 저장·조회·노출 경로 회귀."""

from __future__ import annotations

import json
import unittest
from unittest.mock import AsyncMock, patch

from . import session_persistence
from .contracts.client_affect import (
    CLIENT_AFFECT_DIMENSIONS,
    ClientAffectContextV1,
    ClientAffectDimensionTraceV1,
    ClientAffectPolicyV1,
    ClientAffectTraceV1,
    ClientInnerDisplayV1,
    ClientInnerFeelingV1,
    ClientInnerReactionV1,
    ClientInnerStanceV1,
)
from .deps import Principal, Role
from .paths import repo_path
from .routes import sessions
from .routes import voice as voice_routes
from .services import inner_reaction_exposure, orchestrator
from .services import persona as persona_service
from .services import state_machine
from .services.voice import TTSChunk, VoicePreset
from .session_read_model import SessionReviewReadInput, build_session_review
from .store import InProcSession, TurnRecord, store


def _inner_reaction(turn_seq: int = 2) -> ClientInnerReactionV1:
    return ClientInnerReactionV1(
        schema_version=1,
        turn_seq=turn_seq,
        experienced=("평가받거나 탓을 듣는 것처럼 느꼈다",),
        feelings=(ClientInnerFeelingV1(label="수치심", intensity="뚜렷한"),),
        stance=ClientInnerStanceV1(code="pull_back", label="한발 물러났다"),
        display=ClientInnerDisplayV1(
            code="covered_by_agreement",
            label="속마음과 달리 겉으로는 수긍하는 말로 덮었다",
        ),
        hidden_gap=True,
    )


def _trace() -> ClientAffectTraceV1:
    return ClientAffectTraceV1(
        schema_version=1,
        provider="typesafe",
        model="jev-test",
        latency_ms=10,
        input_tokens=1,
        output_tokens=1,
        cost_usd=None,
        turn_seq=2,
        policy=ClientAffectPolicyV1(
            version="jev-affect-v1",
            min_confidence=0.65,
            accepted_alpha=0.35,
            accepted_cap=0.15,
            tentative_alpha=0.15,
            tentative_cap=0.075,
            tentative_confidence_floor=0.35,
            adjacent_probability_threshold=0.005,
        ),
        context=ClientAffectContextV1(
            stage="라포",
            resistance=0.5,
            effective_openness=0.2,
            rapport_credit=1.0,
        ),
        dimensions=tuple(
            ClientAffectDimensionTraceV1(
                key=dimension,
                before=0.5,
                target=None,
                after=0.5,
                confidence=None,
                probabilities=None,
                decision="held",
            )
            for dimension in CLIENT_AFFECT_DIMENSIONS
        ),
    )


class _Transaction:
    def __init__(self) -> None:
        self.error: type[BaseException] | None = None

    async def __aenter__(self) -> None:
        return None

    async def __aexit__(self, exc_type, exc, tb) -> bool:
        self.error = exc_type
        return False


class _Connection:
    def __init__(self, *, fail_inner_reaction_insert: bool = False) -> None:
        self.fail_inner_reaction_insert = fail_inner_reaction_insert
        self.transaction_context = _Transaction()
        self.executed: list[tuple[str, tuple[object, ...]]] = []

    def transaction(self) -> _Transaction:
        return self.transaction_context

    async def fetchval(self, query: str, *args: object) -> object:
        if "FROM app.sessions" in query:
            return "00000000-0000-0000-0000-000000000111"
        if "COALESCE(MAX(seq)" in query:
            return 2
        if "INSERT INTO app.turns" in query:
            return "00000000-0000-0000-0000-000000000222"
        raise AssertionError(f"unexpected query: {query}")

    async def execute(self, query: str, *args: object) -> str:
        self.executed.append((query, args))
        if self.fail_inner_reaction_insert and "INSERT INTO app.client_inner_reaction" in query:
            raise RuntimeError("inner reaction insert failed")
        return "INSERT 0 1"


class _Acquire:
    def __init__(self, conn: _Connection) -> None:
        self.conn = conn

    async def __aenter__(self) -> _Connection:
        return self.conn

    async def __aexit__(self, exc_type, exc, tb) -> bool:
        return False


class ClientInnerReactionPersistenceTest(unittest.IsolatedAsyncioTestCase):
    async def test_atomic_write_inserts_trace_then_inner_reaction_in_same_transaction(
        self,
    ) -> None:
        conn = _Connection()
        turn = TurnRecord(
            turn_seq=2,
            speaker="client",
            stage="라포",
            text="조금 더 이야기해볼게요.",
            text_masked="조금 더 이야기해볼게요.",
        )
        state = state_machine.SessionState(turn_seq=2)
        reaction = _inner_reaction()

        with (
            patch.object(session_persistence, "get_pool", return_value=object()),
            patch.object(session_persistence, "acquire", return_value=_Acquire(conn)),
        ):
            stored = await session_persistence.append_client_turn_with_affect_trace(
                session_id="00000000-0000-0000-0000-000000000111",
                learner_id="00000000-0000-0000-0000-000000000101",
                turn=turn,
                state=state,
                trace=_trace(),
                inner_reaction=reaction,
            )

        self.assertTrue(stored)
        self.assertEqual(turn.turn_id, "00000000-0000-0000-0000-000000000222")
        queries = [query for query, _args in conn.executed]
        self.assertEqual(
            [q.split()[0:3] for q in queries if "INSERT INTO app.client" in q],
            [
                ["INSERT", "INTO", "app.client_affect_trace"],
                ["INSERT", "INTO", "app.client_inner_reaction"],
            ],
        )
        inner_reaction_query, inner_reaction_args = next(
            (query, args)
            for query, args in conn.executed
            if "INSERT INTO app.client_inner_reaction" in query
        )
        self.assertEqual(
            inner_reaction_args,
            (
                "00000000-0000-0000-0000-000000000222",
                "00000000-0000-0000-0000-000000000111",
                reaction.model_dump(mode="json"),
            ),
        )

    async def test_inner_reaction_none_skips_insert(self) -> None:
        conn = _Connection()
        turn = TurnRecord(
            turn_seq=2,
            speaker="client",
            stage="라포",
            text="조금 더 이야기해볼게요.",
            text_masked="조금 더 이야기해볼게요.",
        )
        state = state_machine.SessionState(turn_seq=2)

        with (
            patch.object(session_persistence, "get_pool", return_value=object()),
            patch.object(session_persistence, "acquire", return_value=_Acquire(conn)),
        ):
            stored = await session_persistence.append_client_turn_with_affect_trace(
                session_id="00000000-0000-0000-0000-000000000111",
                learner_id="00000000-0000-0000-0000-000000000101",
                turn=turn,
                state=state,
                trace=_trace(),
                inner_reaction=None,
            )

        self.assertTrue(stored)
        self.assertFalse(
            any("INSERT INTO app.client_inner_reaction" in query for query, _ in conn.executed)
        )

    async def test_inner_reaction_insert_failure_rolls_back_entire_turn(self) -> None:
        conn = _Connection(fail_inner_reaction_insert=True)
        turn = TurnRecord(
            turn_seq=2,
            speaker="client",
            stage="라포",
            text="조금 더 이야기해볼게요.",
            text_masked="조금 더 이야기해볼게요.",
        )
        state = state_machine.SessionState(turn_seq=2)

        with (
            patch.object(session_persistence, "get_pool", return_value=object()),
            patch.object(session_persistence, "acquire", return_value=_Acquire(conn)),
        ):
            with self.assertRaises(session_persistence.ClientAffectTracePersistenceError):
                await session_persistence.append_client_turn_with_affect_trace(
                    session_id="00000000-0000-0000-0000-000000000111",
                    learner_id="00000000-0000-0000-0000-000000000101",
                    turn=turn,
                    state=state,
                    trace=_trace(),
                    inner_reaction=_inner_reaction(),
                )

        self.assertIsNone(turn.turn_id)
        self.assertIs(conn.transaction_context.error, RuntimeError)


class ListClientInnerReactionTest(unittest.IsolatedAsyncioTestCase):
    async def test_reads_with_caller_role_user_and_cohort_ids(self) -> None:
        reaction = _inner_reaction()
        teacher_principal = Principal(
            user_id="00000000-0000-0000-0000-000000000901",
            role=Role.TEACHER,
            cohort_ids=["cohort-a"],
            email="teacher@hs.ac.kr",
            display_name="Teacher",
            consent_at=1.0,
            profile_completed_at=1.0,
        )

        class FakeConn:
            async def fetch(self, query: str, *args: object):
                self.query = query
                self.args = args
                return [
                    {
                        "turn_id": "00000000-0000-0000-0000-000000000222",
                        "reaction": reaction.model_dump(mode="json"),
                    }
                ]

        conn = FakeConn()

        with (
            patch.object(session_persistence, "get_pool", return_value=object()),
            patch.object(
                session_persistence, "acquire", return_value=_Acquire(conn)
            ) as acquire_mock,
        ):
            result = await session_persistence.list_client_inner_reactions(
                "00000000-0000-0000-0000-000000000111",
                teacher_principal,
            )

        acquire_mock.assert_called_once_with(
            role="teacher",
            user_id="00000000-0000-0000-0000-000000000901",
            cohort_ids=["cohort-a"],
        )
        self.assertIn("00000000-0000-0000-0000-000000000222", result)
        self.assertEqual(
            result["00000000-0000-0000-0000-000000000222"].stance.code, "pull_back"
        )

    async def test_db_failure_falls_back_to_empty_dict_in_dev(self) -> None:
        learner_principal = Principal(
            user_id="00000000-0000-0000-0000-000000000101",
            role=Role.LEARNER,
            cohort_ids=[],
            email="learner@hs.ac.kr",
            display_name="Learner",
            consent_at=1.0,
            profile_completed_at=1.0,
        )
        with patch.object(
            session_persistence, "get_pool", side_effect=RuntimeError("no pool")
        ):
            result = await session_persistence.list_client_inner_reactions(
                "00000000-0000-0000-0000-000000000111",
                learner_principal,
            )

        self.assertEqual(result, {})


class InnerReactionExposureTest(unittest.TestCase):
    def test_returns_none_when_reaction_missing(self) -> None:
        self.assertIsNone(
            inner_reaction_exposure.expose_client_inner_reaction(
                None, stored=True, feedback_enabled=True
            )
        )

    def test_returns_none_when_not_stored(self) -> None:
        self.assertIsNone(
            inner_reaction_exposure.expose_client_inner_reaction(
                _inner_reaction(), stored=False, feedback_enabled=True
            )
        )

    def test_returns_none_when_feedback_policy_disabled(self) -> None:
        self.assertIsNone(
            inner_reaction_exposure.expose_client_inner_reaction(
                _inner_reaction(), stored=True, feedback_enabled=False
            )
        )

    def test_returns_reaction_when_stored_and_policy_enabled(self) -> None:
        reaction = _inner_reaction()
        exposed = inner_reaction_exposure.expose_client_inner_reaction(
            reaction, stored=True, feedback_enabled=True
        )
        self.assertIs(exposed, reaction)


class ClientInnerReactionMigrationSqlTest(unittest.TestCase):
    def setUp(self) -> None:
        self.sql = repo_path(
            "infra", "db", "init", "24_client_inner_reaction.sql"
        ).read_text(encoding="utf-8")

    def test_has_exactly_select_and_insert_policies(self) -> None:
        self.assertIn(
            "CREATE POLICY p_client_inner_reaction_select", self.sql
        )
        self.assertIn(
            "CREATE POLICY p_client_inner_reaction_insert_learner", self.sql
        )
        self.assertEqual(self.sql.count("CREATE POLICY"), 2)

    def test_has_no_update_or_delete_policy(self) -> None:
        self.assertNotIn("FOR UPDATE", self.sql)
        self.assertNotIn("FOR DELETE", self.sql)

    def test_select_policy_blocks_ai_context(self) -> None:
        self.assertIn("NOT app.is_ai_context()", self.sql)

    def test_select_policy_allows_admin_instructor_or_session_owner(self) -> None:
        self.assertIn(
            "app.current_role_name() IN ('admin', 'instructor')", self.sql
        )
        self.assertIn("s.learner_id = app.current_uid()", self.sql)


def _principal(*, learner_feedback_enabled: bool = True) -> Principal:
    return Principal(
        user_id="00000000-0000-0000-0000-000000000501",
        role=Role.LEARNER,
        cohort_ids=[],
        email="inner-reaction-route-test@hs.ac.kr",
        display_name="Inner Reaction Route Test",
        consent_at=1.0,
        profile_completed_at=1.0,
        learner_feedback_enabled=learner_feedback_enabled,
    )


def _session(principal: Principal, session_id: str) -> InProcSession:
    card = persona_service.P1
    sess = InProcSession(
        session_id=session_id,
        case_id=f"{session_id}-case",
        learner_id=principal.user_id,
        persona_code=card.code,
        theory_mode="humanistic",
        persona=card,
        state=state_machine.SessionState(
            resistance=card.base_resistance(),
            ideation_stage=card.ideation_baseline(),
        ),
    )
    store.put(sess)
    return sess


async def _successful_turn_with_reaction(reaction, ctx, engine, **kwargs):
    assert ctx.state_after is not None
    ctx.client_affect_trace = _trace()
    ctx.client_inner_reaction = reaction
    return orchestrator.TurnResult(
        turn_seq=ctx.state_after.turn_seq,
        stage=ctx.state_after.stage.value,
        effective_openness=ctx.state_after.effective_openness,
        client_reply="조금 더 말해볼게요.",
        safety_flagged=False,
        state_after=ctx.state_after,
    )


class RouteInnerReactionParityTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        store._sessions.clear()
        sessions._RECALL_CACHE.clear()

    async def asyncTearDown(self) -> None:
        store._sessions.clear()
        sessions._RECALL_CACHE.clear()

    async def test_submit_turn_includes_inner_reaction_when_policy_enabled(self) -> None:
        principal = _principal(learner_feedback_enabled=True)
        sess = _session(principal, "inner-reaction-turn-on")
        reaction = _inner_reaction()

        async def successful_turn(ctx, engine, **kwargs):
            return await _successful_turn_with_reaction(reaction, ctx, engine, **kwargs)

        with (
            patch.object(sessions.orchestrator, "run_turn_generate", successful_turn),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
        ):
            response = await sessions.submit_turn(
                sess.session_id,
                sessions.TurnRequest(text="속마음 노출 테스트"),
                principal,
            )

        self.assertIsNotNone(response.inner_reaction)
        self.assertTrue(response.inner_reaction.hidden_gap)
        self.assertEqual(response.inner_reaction.stance.code, "pull_back")

    async def test_submit_turn_omits_inner_reaction_when_policy_disabled(self) -> None:
        principal = _principal(learner_feedback_enabled=False)
        sess = _session(principal, "inner-reaction-turn-off")
        reaction = _inner_reaction()

        async def successful_turn(ctx, engine, **kwargs):
            return await _successful_turn_with_reaction(reaction, ctx, engine, **kwargs)

        with (
            patch.object(sessions.orchestrator, "run_turn_generate", successful_turn),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
        ):
            response = await sessions.submit_turn(
                sess.session_id,
                sessions.TurnRequest(text="속마음 비노출 테스트"),
                principal,
            )

        self.assertIsNone(response.inner_reaction)

    async def _stream_done_payload(self, response: object) -> dict:
        async for chunk in response.body_iterator:  # type: ignore[attr-defined]
            if isinstance(chunk, dict) and chunk.get("event") == "done":
                return json.loads(chunk["data"])
        raise AssertionError("done event not found in stream")

    async def test_stream_turn_done_includes_inner_reaction_when_policy_enabled(
        self,
    ) -> None:
        principal = _principal(learner_feedback_enabled=True)
        sess = _session(principal, "inner-reaction-stream-on")
        reaction = _inner_reaction()

        async def successful_stream(ctx, engine, **kwargs):
            assert ctx.state_after is not None
            ctx.client_affect_trace = _trace()
            ctx.client_inner_reaction = reaction
            yield orchestrator.StreamEvent("token", {"text": "속마음 스트림 테스트"})
            yield orchestrator.StreamEvent(
                "done",
                {
                    "session_id": ctx.session_id,
                    "stage": ctx.state_after.stage.value,
                    "effective_openness": ctx.state_after.effective_openness,
                    "turn_seq": ctx.state_after.turn_seq,
                    "safety_flagged": False,
                },
            )

        with (
            patch.object(sessions.orchestrator, "run_turn_stream", successful_stream),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
        ):
            response = await sessions.stream_turn(
                sess.session_id,
                sessions.TurnRequest(text="속마음 스트림 발화"),
                principal,
            )
            done_payload = await self._stream_done_payload(response)

        self.assertIsNotNone(done_payload.get("inner_reaction"))
        self.assertEqual(done_payload["inner_reaction"]["stance"]["code"], "pull_back")

    async def test_stream_turn_done_omits_inner_reaction_when_policy_disabled(
        self,
    ) -> None:
        principal = _principal(learner_feedback_enabled=False)
        sess = _session(principal, "inner-reaction-stream-off")
        reaction = _inner_reaction()

        async def successful_stream(ctx, engine, **kwargs):
            assert ctx.state_after is not None
            ctx.client_affect_trace = _trace()
            ctx.client_inner_reaction = reaction
            yield orchestrator.StreamEvent("token", {"text": "속마음 스트림 비노출 테스트"})
            yield orchestrator.StreamEvent(
                "done",
                {
                    "session_id": ctx.session_id,
                    "stage": ctx.state_after.stage.value,
                    "effective_openness": ctx.state_after.effective_openness,
                    "turn_seq": ctx.state_after.turn_seq,
                    "safety_flagged": False,
                },
            )

        with (
            patch.object(sessions.orchestrator, "run_turn_stream", successful_stream),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
        ):
            response = await sessions.stream_turn(
                sess.session_id,
                sessions.TurnRequest(text="속마음 스트림 비노출 발화"),
                principal,
            )
            done_payload = await self._stream_done_payload(response)

        self.assertIsNone(done_payload.get("inner_reaction"))

    async def test_voice_reply_includes_inner_reaction_when_policy_enabled(self) -> None:
        principal = _principal(learner_feedback_enabled=True)
        sess = _session(principal, "inner-reaction-voice-on")
        reaction = _inner_reaction()

        class FakeWebSocket:
            def __init__(self) -> None:
                self.messages: list[dict[str, object]] = []
                self.client_state = voice_routes.WebSocketState.CONNECTED

            async def send_text(self, data: str) -> None:
                self.messages.append(json.loads(data))

            async def send_bytes(self, data: bytes) -> None:
                pass

        async def successful_turn(ctx, engine, **kwargs):
            return await _successful_turn_with_reaction(reaction, ctx, engine, **kwargs)

        async def fake_synthesize_stream(text, voice_preset):
            yield TTSChunk(audio=b"tts-audio")

        websocket = FakeWebSocket()
        with (
            patch.object(voice_routes.orchestrator, "run_turn_generate", successful_turn),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
            patch.object(
                voice_routes.voice_service, "synthesize_stream", fake_synthesize_stream
            ),
        ):
            await voice_routes._run_turn_and_speak(
                websocket,  # type: ignore[arg-type]
                voice_routes.VoiceSessionContext(
                    session_id=sess.session_id,
                    principal=principal,
                    voice_preset=VoicePreset(preset="neutral", openai_voice="sage"),
                ),
                voice_routes.VoiceTurnInput(learner_text="속마음 노출 음성 발화"),
            )

        reply_message = next(
            message for message in websocket.messages if message.get("type") == "reply"
        )
        self.assertIsNotNone(reply_message.get("inner_reaction"))
        self.assertEqual(reply_message["inner_reaction"]["stance"]["code"], "pull_back")

    async def test_voice_reply_omits_inner_reaction_when_policy_disabled(self) -> None:
        principal = _principal(learner_feedback_enabled=False)
        sess = _session(principal, "inner-reaction-voice-off")
        reaction = _inner_reaction()

        class FakeWebSocket:
            def __init__(self) -> None:
                self.messages: list[dict[str, object]] = []
                self.client_state = voice_routes.WebSocketState.CONNECTED

            async def send_text(self, data: str) -> None:
                self.messages.append(json.loads(data))

            async def send_bytes(self, data: bytes) -> None:
                pass

        async def successful_turn(ctx, engine, **kwargs):
            return await _successful_turn_with_reaction(reaction, ctx, engine, **kwargs)

        async def fake_synthesize_stream(text, voice_preset):
            yield TTSChunk(audio=b"tts-audio")

        websocket = FakeWebSocket()
        with (
            patch.object(voice_routes.orchestrator, "run_turn_generate", successful_turn),
            patch.object(
                session_persistence,
                "append_client_turn_with_affect_trace",
                AsyncMock(return_value=True),
            ),
            patch.object(
                voice_routes.voice_service, "synthesize_stream", fake_synthesize_stream
            ),
        ):
            await voice_routes._run_turn_and_speak(
                websocket,  # type: ignore[arg-type]
                voice_routes.VoiceSessionContext(
                    session_id=sess.session_id,
                    principal=principal,
                    voice_preset=VoicePreset(preset="neutral", openai_voice="sage"),
                ),
                voice_routes.VoiceTurnInput(learner_text="속마음 비노출 음성 발화"),
            )

        reply_message = next(
            message for message in websocket.messages if message.get("type") == "reply"
        )
        self.assertIsNone(reply_message.get("inner_reaction"))


def _review_session(*, learner_feedback_enabled: bool) -> InProcSession:
    state = state_machine.init_state(params=persona_service.P1.openness_params())
    return InProcSession(
        session_id="00000000-0000-4000-8000-000000000402",
        case_id="inner-reaction-review-case",
        learner_id="00000000-0000-0000-0000-000000000402",
        persona_code=persona_service.P1.code,
        theory_mode="humanistic",
        persona=persona_service.P1,
        state=state,
        created_at=1_000.0,
        ended_at=1_120.0,
        ended=True,
        learner_feedback_enabled=learner_feedback_enabled,
        turns=[
            TurnRecord(
                turn_seq=1,
                speaker="counselor",
                stage=state.stage.value,
                text="상담자 발화",
                text_masked="상담자 발화",
                turn_id="00000000-0000-0000-0000-000000000501",
            ),
            TurnRecord(
                turn_seq=2,
                speaker="client",
                stage=state.stage.value,
                text="가상 내담자 응답",
                text_masked="가상 내담자 응답",
                turn_id="00000000-0000-0000-0000-000000000502",
            ),
        ],
    )


class ReviewInnerReactionMappingTest(unittest.TestCase):
    def test_client_turn_gets_inner_reaction_mapped_by_turn_id(self) -> None:
        sess = _review_session(learner_feedback_enabled=True)
        reaction = _inner_reaction()

        review = build_session_review(
            SessionReviewReadInput(
                session=sess,
                evaluation_record={"status": "ready", "payload": {"summary": "AI 요약"}},
                evaluation_durable=True,
                learner_feedback_enabled=True,
                expose_learner_feedback=True,
                inner_reactions={"00000000-0000-0000-0000-000000000502": reaction},
                now_ts=1_120.0,
            )
        )

        learner_turn, client_turn = review.turns
        self.assertIsNone(learner_turn.innerReaction)
        self.assertIsNotNone(client_turn.innerReaction)
        self.assertEqual(client_turn.innerReaction.stance.code, "pull_back")

    def test_feedback_hidden_omits_inner_reaction_even_when_present(self) -> None:
        sess = _review_session(learner_feedback_enabled=False)
        reaction = _inner_reaction()

        review = build_session_review(
            SessionReviewReadInput(
                session=sess,
                learner_feedback_enabled=False,
                expose_learner_feedback=False,
                inner_reactions={"00000000-0000-0000-0000-000000000502": reaction},
                now_ts=1_120.0,
            )
        )

        self.assertTrue(all(turn.innerReaction is None for turn in review.turns))


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