"""Round two of the review: the administrator's surface, and four edges.

Each of these is either something an employee is told that is not true, or
something the console cannot see that it exists to show.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant import admin_service
from app.assistant.actions import ACTION_SUPPORT_TICKET, ACTION_VPN_APPLICATION
from app.assistant.models import (
    CHANNEL_FEISHU,
    ROLE_ASSISTANT,
    ROLE_USER,
    AssistantConversation,
    AssistantMessage,
)
from tests.test_assistant.conftest import make_feishu_person


class TestTheTranscriptCannotInterleave:
    """``seq`` orders the transcript, and its guard fails open.

    The Redis turn lock and the event-dedup claim BOTH fail open, so a Redis
    outage lets two concurrent turns read the same ``turn_count`` and write the
    same pair of sequences. Only a non-unique index stood between that and a
    silently interleaved transcript with nothing able to say which reply answered
    which question. Verified against production before adding the constraint:
    78 rows, zero duplicate pairs.
    """

    @pytest.mark.asyncio
    async def test_the_database_refuses_a_duplicate_sequence(
        self, db_session: AsyncSession, feishu_person,
    ) -> None:
        from sqlalchemy.exc import IntegrityError

        conversation = AssistantConversation(
            person_id=feishu_person.id, channel=CHANNEL_FEISHU, chat_id="oc-seq",
        )
        db_session.add(conversation)
        await db_session.flush()
        db_session.add(
            AssistantMessage(
                conversation_id=conversation.id, seq=0, role=ROLE_USER, content="q",
            ),
        )
        await db_session.commit()

        db_session.add(
            AssistantMessage(
                conversation_id=conversation.id, seq=0, role=ROLE_USER, content="q2",
            ),
        )
        with pytest.raises(IntegrityError):
            await db_session.commit()
        await db_session.rollback()

    @pytest.mark.asyncio
    async def test_two_conversations_may_share_a_sequence(
        self, db_session: AsyncSession, feishu_person,
    ) -> None:
        """The constraint is per-thread, not global."""
        ids = []
        for chat in ("oc-a", "oc-b"):
            conversation = AssistantConversation(
                person_id=feishu_person.id, channel=CHANNEL_FEISHU, chat_id=chat,
            )
            db_session.add(conversation)
            await db_session.flush()
            ids.append(conversation.id)
        for conversation_id in ids:
            db_session.add(
                AssistantMessage(
                    conversation_id=conversation_id, seq=0,
                    role=ROLE_USER, content="q",
                ),
            )
        await db_session.commit()


class TestARefusedTapIsCountedSomewhere:
    """Every refused tap is audited with ``status="denied"`` — and nothing read
    those rows, so the forwarded-card impersonation the write path exists to
    block was invisible on the page whose docstring calls a refusal a
    first-class number."""

    @pytest.mark.asyncio
    async def test_a_refused_tap_shows_up_in_the_stats(
        self, db_session: AsyncSession, cards, enabled_gate,
    ) -> None:
        from app.assistant.card_actions import ACTION_CONFIRM, handle_card_action

        owner = await make_feishu_person(
            db_session, subject="on_owner_0001", email="owner@test.com",
        )
        # Someone else's proposal, tapped from a forwarded card.
        from app.assistant.models import STATUS_PENDING, AssistantActionProposal

        proposal = AssistantActionProposal(
            person_id=owner.id,
            action_key=ACTION_SUPPORT_TICKET,
            params={"title": "t", "body": "b"},
            params_hash="d",
            status=STATUS_PENDING,
            expires_at=datetime.now(UTC) + timedelta(minutes=30),
        )
        db_session.add(proposal)
        colleague = await make_feishu_person(
            db_session, subject="on_colleague_0001", email="colleague@test.com",
        )
        await db_session.commit()
        assert colleague.id != owner.id

        before = (await admin_service.usage_stats(db_session, days=1))[
            "refused_card_taps"
        ]
        result = await handle_card_action(
            {
                "header": {"event_type": "card.action.trigger"},
                "event": {
                    "operator": {
                        "open_id": "ou_colleague_0001",
                        "union_id": "on_colleague_0001",
                    },
                    "action": {"value": {
                        "action": ACTION_CONFIRM, "proposal": str(proposal.id),
                    }},
                },
            },
            trace_id=str(uuid.uuid4()),
        )

        assert result["toast"]["type"] == "error"
        await db_session.rollback()
        after = (await admin_service.usage_stats(db_session, days=1))[
            "refused_card_taps"
        ]
        assert after == before + 1, "the refusal was audited but not counted"


class TestTheAuditWindowMeansWhatTheCallerSaid:
    """The driver formats a datetime and IGNORES tzinfo, while these columns hold
    UTC wall clock — so ``since=…+08:00`` silently shifted the window by eight
    hours, with conversations missing from the edges of the range asked for."""

    @pytest.mark.asyncio
    async def test_an_offset_aware_bound_is_converted_not_dropped(
        self, db_session: AsyncSession, feishu_person,
    ) -> None:
        from zoneinfo import ZoneInfo

        # A conversation four hours ago in UTC terms.
        four_hours_ago = datetime.now(UTC) - timedelta(hours=4)
        conversation = AssistantConversation(
            person_id=feishu_person.id, channel=CHANNEL_FEISHU, chat_id="oc-tz",
            last_message_at=four_hours_ago.replace(tzinfo=None),
        )
        db_session.add(conversation)
        await db_session.commit()

        # The same instant expressed in +08:00, minus a minute, must include it.
        bound = (four_hours_ago - timedelta(minutes=1)).astimezone(
            ZoneInfo("Asia/Shanghai"),
        )
        rows, total = await admin_service.list_conversations(
            db_session,
            filters=admin_service.ConversationFilters(
                person_id=feishu_person.id, since=bound,
            ),
        )
        assert total == 1, (
            "an offset-aware bound was read as naive and moved the window"
        )
        assert str(rows[0]["id"]) == str(conversation.id)


class TestAJoinerIsNotOfferedSomethingThatCannotRun:
    """Read access is wider than write access on purpose. A pre-hire whose Feishu
    account is already open may ASK — but every domain service enforces
    ``is_active_employee``, so a card offered to them would be approved and then
    refused at execute time, leaving a red card and no explanation."""

    @pytest.mark.asyncio
    async def test_a_pre_hire_gets_an_explanation_not_a_card(
        self, db_session: AsyncSession, cards,
    ) -> None:
        from app.assistant.propose_tools import make_dispatch

        joiner = await make_feishu_person(
            db_session, subject="on_joiner_0001", email="joiner@test.com",
            employment_status="pre_hire",
        )
        dispatch = make_dispatch(
            person=joiner, chat_id="oc-joiner", open_id="ou_joiner_0001",
            lang="zh", trace_id=str(uuid.uuid4()),
        )

        out = await dispatch(
            f"propose_{ACTION_VPN_APPLICATION}",
            {
                "vpn_node": "CN4-HK",
                "usage_description": "需要访问内网 Grafana",
                "resource_checklist": "grafana.internal",
            },
            db_session,
            joiner,
        )

        import json

        payload = json.loads(out)
        assert payload["error"] == "not_an_active_employee"
        assert payload["employment_status"] == "pre_hire"
        assert cards == [], "a card that cannot execute was sent anyway"

    @pytest.mark.asyncio
    async def test_an_active_employee_is_unaffected(
        self, db_session: AsyncSession, feishu_person, cards,
    ) -> None:
        from app.assistant.propose_tools import make_dispatch

        dispatch = make_dispatch(
            person=feishu_person, chat_id="oc-active", open_id="ou-active",
            lang="zh", trace_id=str(uuid.uuid4()),
        )
        out = await dispatch(
            f"propose_{ACTION_VPN_APPLICATION}",
            {
                "vpn_node": "CN4-HK",
                "usage_description": "需要访问内网 Grafana",
                "resource_checklist": "grafana.internal",
            },
            db_session,
            feishu_person,
        )
        assert "awaiting_confirmation" in out
        assert len(cards) == 1


class TestTheBusyReplyDoesNotPromiseAQueue:
    """There is no queue. The message is dropped, and its event id is already in
    a 6-hour dedup key, so Feishu's own redelivery is discarded too."""

    def test_it_asks_them_to_resend_in_every_language(self) -> None:
        from app.assistant import messages as canned

        for lang, promise in (
            ("zh", "马上看这条"), ("en", "get to this next"), ("ja", "次にこちらを確認"),
        ):
            text = canned.busy(lang)
            assert promise not in text, f"{lang}: still promises a queue"
            assert text.strip(), f"{lang}: says nothing"

    def test_the_stats_scan_is_bounded(self) -> None:
        """A stats endpoint the console calls on open, over a window up to a
        year, for a top-10 list."""
        assert admin_service._TOOL_SCAN_LIMIT > 0  # noqa: SLF001


class TestTheReviewRoundsOwnRegressions:
    """Every one of these is a defect a fix in this same batch introduced.

    They are grouped together deliberately: the pattern worth defending against
    is not any one of them but "the repair broke the thing next to it".
    """

    @pytest.mark.asyncio
    async def test_a_seq_collision_costs_the_rating_buttons_not_the_answer(
        self, db_session: AsyncSession, feishu_person, kernel, sent, enabled_gate,
    ) -> None:
        """The unique constraint turned a Redis-outage misordering into a turn
        that reached the employee not at all — the exact failure the round
        before it fixed from a different cause."""
        from app.assistant import conversation
        from app.assistant import session as session_store
        from app.assistant.models import AssistantConversation

        # A thread whose next sequence pair is already taken while its
        # ``turn_count`` still reads 0 — exactly the state two concurrent turns
        # produce when the Redis lock failed open.
        thread = AssistantConversation(
            person_id=feishu_person.id, channel=CHANNEL_FEISHU, chat_id="oc-clash",
            turn_count=0,
        )
        db_session.add(thread)
        await db_session.flush()
        for seq in (0, 1, 2, 3):
            db_session.add(
                AssistantMessage(
                    conversation_id=thread.id, seq=seq,
                    role=ROLE_USER if seq % 2 == 0 else ROLE_ASSISTANT,
                    content=f"taken {seq}",
                ),
            )
        await db_session.commit()
        # Working memory is what makes the turn CONTINUE this thread rather than
        # open a fresh one — without it the collision never happens and the test
        # passes with the retry removed, which is how its first cut proved
        # nothing.
        await session_store.save_history(
            db_session, person_id=feishu_person.id, chat_id="oc-clash",
            history=[{"role": "user", "content": "先前的问题"}],
        )

        # Read BEFORE the turn: the retry path rolls back, which expires every
        # instance in this session — including this one.
        thread_id = thread.id

        turn = await conversation.run_turn(
            db_session,
            person=feishu_person,
            chat_id="oc-clash",
            question="我的电脑保修到什么时候",
            lang="zh",
            trace_id=str(uuid.uuid4()),
        )

        # The answer survives, which is the whole point.
        assert turn.outcome == "answered"
        assert turn.reply.strip()
        # It either found a free pair on retry, or gave up on the transcript —
        # never on the employee.
        await db_session.rollback()
        rows = (
            await db_session.execute(
                select(AssistantMessage).where(
                    AssistantMessage.conversation_id == thread_id,
                ),
            )
        ).scalars().all()
        assert len(rows) >= 4

    @pytest.mark.asyncio
    async def test_an_epoch_outlives_the_history_it_guards(
        self, db_session: AsyncSession, feishu_person,
    ) -> None:
        """Asserted on the KEY's real TTL in Redis, not on the constant.

        A guard that expires before the thread it guards leaves two holes: a turn
        whose captured epoch has since vanished is refused for nothing, and a
        second reset re-creating the key from zero hands out a value an in-flight
        turn already holds (``INCR`` on a missing key returns 1 again), waving its
        stale write through.
        """
        from app.assistant import session as session_store
        from app.core.redis_client import get_redis

        chat_id = f"oc-ttl-{uuid.uuid4().hex[:8]}"
        await session_store.save_history(
            db_session, person_id=feishu_person.id, chat_id=chat_id,
            history=[{"role": "user", "content": "x"}],
        )
        assert await session_store.clear_history(
            person_id=feishu_person.id, chat_id=chat_id,
        )

        redis = get_redis()
        epoch_ttl = await redis.ttl(
            session_store._epoch_key(feishu_person.id, chat_id),  # noqa: SLF001
        )
        # Both bounds: the configured default AND the ceiling the setting can be
        # raised to. Dropping the link to ``DEFAULT_SESSION_TTL_MINUTES`` would
        # let this assertion stop tracking a raised session TTL.
        from app.assistant.settings import DEFAULT_SESSION_TTL_MINUTES

        history_ttl_ceiling = 1440 * 60  # ``bounded_int_setting``'s maximum
        assert epoch_ttl > DEFAULT_SESSION_TTL_MINUTES * 60
        assert epoch_ttl > history_ttl_ceiling, (
            f"the epoch expires in {epoch_ttl}s, before the history it guards"
        )

    @pytest.mark.asyncio
    async def test_a_reset_that_did_not_stick_is_not_announced_as_done(
        self, db_session: AsyncSession, feishu_person, sent, enabled_gate,
        monkeypatch,
    ) -> None:
        """The employee was told 「好的，我们从头开始」 while the clear had failed."""
        from app.assistant import session as session_store
        from app.assistant.feishu_events import handle_message_event
        from tests.test_assistant.conftest import message_event

        async def _fails(**_kw):
            return False

        monkeypatch.setattr(session_store, "clear_history", _fails)

        outcome = await handle_message_event(
            db_session,
            message_event(text="新话题", event_id=f"evt-{uuid.uuid4()}"),
            trace_id=str(uuid.uuid4()),
        )

        assert outcome == "reset_failed"
        assert sent, "the employee was told nothing at all"
        told = sent[-1]["text"]
        assert "从头开始" not in told, "a failed reset was announced as done"
        # And it says the one thing they need: the thread is still there.
        assert "上下文还在" in told, f"unhelpful copy for a failed reset: {told!r}"

    @pytest.mark.asyncio
    async def test_the_escalation_path_applies_the_same_submit_gate(
        self, db_session: AsyncSession, cards, sent,
    ) -> None:
        """转人工 reaches the same domain service, and bypassed the gate — so a
        joiner tapping it got a card, approved it, and met a red 「没能提交」."""
        from app.assistant import escalation
        from tests.test_assistant.conftest import make_feishu_person

        joiner = await make_feishu_person(
            db_session, subject="on_joiner_esc", email="joiner.esc@test.com",
            employment_status="pre_hire",
        )

        outcome = await escalation.offer_ticket(
            db_session,
            person=joiner,
            chat_id="oc-esc",
            open_id="ou_joiner_esc",
            lang="zh",
            trace_id=str(uuid.uuid4()),
        )

        assert outcome == "not_active_employee"
        assert cards == [], "a card that cannot execute was sent anyway"
        assert sent, "and nothing explained why"

    @pytest.mark.asyncio
    async def test_a_nonsense_flow_is_not_credited_as_the_right_one(
        self, db_session: AsyncSession, feishu_person, cards,
    ) -> None:
        """Asserted on the guard's ANSWER, not on the source text.

        Recording every failed validation let a flow the model merely guessed at
        — 「椅子坏了」 matches ``complete_api_key_rotation`` on CJK bigram overlap
        — make the ticket guard reply "you already found the right flow, go and
        ask for the missing value". A source-text assertion could not see that,
        and the very next change to this line kept the substring while widening
        the rule.
        """
        import json

        from app.assistant.propose_tools import make_dispatch

        dispatch = make_dispatch(
            person=feishu_person, chat_id="oc-guess", open_id="ou-guess",
            lang="zh", trace_id=str(uuid.uuid4()),
        )

        # A flow the model was never shown, reached through the generic tool.
        guessed = json.loads(await dispatch(
            "propose_helix_action",
            {"action": "complete_api_key_rotation", "arguments": {}},
            db_session, feishu_person,
        ))
        assert guessed["error"] == "invalid_parameters"

        out = json.loads(await dispatch(
            f"propose_{ACTION_SUPPORT_TICKET}",
            {"title": "工位的椅子坏了", "body": "液压杆不回弹"},
            db_session, feishu_person,
        ))
        assert out["error"] != "the_flow_you_tried_only_needed_a_value", (
            "a guessed flow was credited as the one the employee asked for"
        )
        assert cards == []

    @pytest.mark.asyncio
    async def test_a_derived_flow_named_out_of_thin_air_is_not_credited(
        self, db_session: AsyncSession, feishu_person, cards,
    ) -> None:
        """The gap the previous round's fix left open through a second entrance.

        ``action_for_tool(name) is not None`` was tautological where it was used
        (the branch above already returned for an unresolvable name), and
        ``get_action`` falls back to the derived catalogue — so ANY of the 60
        derived flows named as ``propose_<key>`` counted as "advertised", which is
        the guessed-flow hole in different clothing. Only the three the prompt
        actually names, plus whatever this turn's search surfaced, may count.
        """
        import json

        from app.assistant.propose_tools import make_dispatch

        dispatch = make_dispatch(
            person=feishu_person, chat_id="oc-thin-air", open_id="ou-thin-air",
            lang="zh", trace_id=str(uuid.uuid4()),
        )
        # A derived flow the prompt does not name and no search surfaced.
        guessed = json.loads(await dispatch(
            "propose_complete_api_key_rotation", {}, db_session, feishu_person,
        ))
        assert guessed["error"] == "invalid_parameters"

        out = json.loads(await dispatch(
            f"propose_{ACTION_SUPPORT_TICKET}",
            {"title": "工位的椅子坏了", "body": "液压杆不回弹"},
            db_session, feishu_person,
        ))
        assert out["error"] != "the_flow_you_tried_only_needed_a_value", (
            "a derived flow nobody advertised was credited as the right one"
        )

    @pytest.mark.asyncio
    async def test_a_flow_the_model_was_shown_still_earns_the_branch(
        self, db_session: AsyncSession, feishu_person, cards,
    ) -> None:
        """The other half: a NAMED curated tool is advertised by the prompt, so a
        validation failure on it is real evidence."""
        import json

        from app.assistant.actions import ACTION_ASSET_REQUEST
        from app.assistant.propose_tools import make_dispatch

        dispatch = make_dispatch(
            person=feishu_person, chat_id="oc-named", open_id="ou-named",
            lang="zh", trace_id=str(uuid.uuid4()),
        )
        shown = json.loads(await dispatch(
            f"propose_{ACTION_ASSET_REQUEST}", {"request_type": "self_borrow"},
            db_session, feishu_person,
        ))
        assert shown["error"] == "invalid_parameters"

        out = json.loads(await dispatch(
            f"propose_{ACTION_SUPPORT_TICKET}",
            {"title": "我要借一台显示器", "body": "居家办公"},
            db_session, feishu_person,
        ))
        assert out["error"] == "the_flow_you_tried_only_needed_a_value"
        assert ACTION_ASSET_REQUEST in out["flows_you_already_reached_for"]

    # The multi-batch embed's deadline, concurrency and sibling cancellation are
    # asserted on BEHAVIOUR in ``tests/test_support/test_bailian_embed_batching``
    # — a constant comparison here could not see the leak that review found.

    @pytest.mark.asyncio
    async def test_the_audit_row_survives_a_transcript_retry(
        self, db_session: AsyncSession, feishu_person, kernel, sent, enabled_gate,
    ) -> None:
        """The seq-clash retry rolls back, which expires the caller's ``person``.

        ``feishu_events`` then reads ``person.id`` for its audit row — lazy IO on
        an expired instance, ``MissingGreenlet``, and the record of the turn is
        gone. Same class as the failure-path fix one round earlier, reached
        through the retry added to fix the round after that.
        """
        from app.assistant import session as session_store
        from app.assistant.feishu_events import handle_message_event
        from app.assistant.models import AssistantConversation
        from app.audit.models import AuditLog
        from tests.test_assistant.conftest import message_event

        thread = AssistantConversation(
            person_id=feishu_person.id, channel=CHANNEL_FEISHU, chat_id="oc-0001",
            turn_count=0,
        )
        db_session.add(thread)
        await db_session.flush()
        for seq in (0, 1):
            db_session.add(
                AssistantMessage(
                    conversation_id=thread.id, seq=seq,
                    role=ROLE_USER if seq == 0 else ROLE_ASSISTANT,
                    content=f"taken {seq}",
                ),
            )
        await db_session.commit()
        await session_store.save_history(
            db_session, person_id=feishu_person.id, chat_id="oc-0001",
            history=[{"role": "user", "content": "先前的问题"}],
        )

        trace_id = str(uuid.uuid4())
        outcome = await handle_message_event(
            db_session,
            message_event(text="我的电脑保修到什么时候", event_id=f"evt-{uuid.uuid4()}"),
            trace_id=trace_id,
        )

        assert outcome == "answered"
        assert sent, "the employee was told nothing"
        await db_session.rollback()
        rows = (
            await db_session.execute(
                select(AuditLog).where(
                    AuditLog.action == "assistant.message.received",
                    AuditLog.trace_id == trace_id,
                ),
            )
        ).scalars().all()
        assert len(rows) == 1, "the retry cost the turn its audit row"

    def test_a_named_curated_flow_can_earn_the_rejection_branch(self) -> None:
        """``create_asset_request`` is not in the derived catalogue at all, so
        keying only on this turn's search left one of the three curated flows
        permanently unable to be credited as "the flow you already found"."""
        from app.assistant.actions import ACTION_ASSET_REQUEST
        from app.assistant.derived_actions import derived_catalogue
        from app.assistant.propose_tools import action_for_tool

        assert ACTION_ASSET_REQUEST not in derived_catalogue()
        # …but the prompt names it, and the named tool resolves — which is the
        # evidence the guard now accepts.
        assert action_for_tool(f"propose_{ACTION_ASSET_REQUEST}") is not None