"""Tests for the MAF roundtable V2 Conversation API (Phase 1 / R1).

Covers:
- Create, list, detail, message pagination, user isolation.
- Public API responses never leak internal_session_id, active_run_id, run_id,
  owner_user_id, thread ids, or secret-class fields.
- First message creates a real internal session; run FK is valid.
- Transaction rollback: any flush failure rolls back message + run + event + status.
- Concurrent send: only one succeeds, the other gets 409; no duplicate runs.
- Mixed tool-events and messages: last_stream_seq is contiguous and usable as SSE cursor.
- Empty / overlong / forged-internal-field content → 422.
- Non-owner cannot read/write; admin cannot write another user's conversation.
"""

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import pytest
import pytest_asyncio
from sqlalchemy import inspect, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.routers.maf_conversations import _user_ctx
from app.maf_roundtable.conversation_schemas import (
    CreateConversationRequest,
    SendMessageRequest,
)
from app.maf_roundtable.conversation_service import (
    ConversationConflictError,
    ConversationDisabledError,
    ConversationNotFoundError,
    MafConversationService,
)
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import (
    MafRoundtableConversationRow,
    MafRoundtableMessageRow,
    MafRoundtableRunRow,
    MafRoundtableSessionRow,
)


class FakeMafConfig:
    enabled = True
    default_mode = "structured"
    default_model_name = "test-model"
    default_reasoning_effort = None
    default_manager_agent_id = "mgr-agent"
    default_max_rounds = 8
    default_write_policy = "confirm_high_risk"
    admin_diagnostic_access = True


@pytest_asyncio.fixture
async def store(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'conv.db'}")
    async with engine.begin() as conn:
        import deerflow.persistence.maf_roundtable.model  # noqa: F401

        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    yield make_maf_roundtable_store(sf)
    await engine.dispose()


@pytest_asyncio.fixture
async def svc(store):
    return MafConversationService(store, maf_config=FakeMafConfig())


FORBIDDEN_FIELDS = {
    "owner_user_id", "internal_session_id", "active_run_id",
    "thread_id", "capability_snapshot", "idempotency_key",
    "target_key", "internal_canary",
}


@pytest.mark.asyncio
async def test_create_and_get_conversation(svc):
    conv = await svc.create_conversation(user_id="u1", title="Test", task_context={"business_code": "3Q"})
    assert conv["id"]
    assert conv["title"] == "Test"
    assert conv["safe_status"] == "idle"
    assert conv["last_stream_seq"] == 0
    assert conv["task_context"] == {"business_code": "3Q"}
    for field in FORBIDDEN_FIELDS:
        assert field not in conv, f"Leaked field: {field}"

    fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="u1")
    assert fetched["id"] == conv["id"]
    for field in FORBIDDEN_FIELDS:
        assert field not in fetched


@pytest.mark.asyncio
async def test_list_conversations_owner_scoped(svc):
    await svc.create_conversation(user_id="u1", title="A")
    await svc.create_conversation(user_id="u1", title="B")
    await svc.create_conversation(user_id="u2", title="C")

    list_u1, total_u1 = await svc.list_conversations(user_id="u1")
    assert total_u1 == 2
    assert all(c["id"] for c in list_u1)

    list_u2, total_u2 = await svc.list_conversations(user_id="u2")
    assert total_u2 == 1


@pytest.mark.asyncio
async def test_get_conversation_wrong_owner_returns_404(svc):
    conv = await svc.create_conversation(user_id="u1", title="Private")
    with pytest.raises(ConversationNotFoundError):
        await svc.get_conversation(conversation_id=conv["id"], user_id="u2")


@pytest.mark.asyncio
async def test_admin_can_read_but_not_write(svc):
    conv = await svc.create_conversation(user_id="u1", title="AdminRead")
    fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="admin", is_admin=True)
    assert fetched["id"] == conv["id"]


@pytest.mark.asyncio
async def test_send_message_creates_real_session_and_run(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="SendTest")
    result = await svc.send_message(conversation_id=conv["id"], user_id="u1", content="Hello world")

    assert result["message_id"]
    assert result["conversation_id"] == conv["id"]
    assert result["seq"] == 1

    async with store.session_factory() as session:
        conv_row = (await session.execute(
            select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
        )).scalars().first()
        assert conv_row is not None
        assert conv_row.internal_session_id is not None
        assert conv_row.safe_status == "running"
        assert conv_row.active_run_id is not None
        # send_message writes only user.message.created; the manager.dispatch
        # projection comes later from the real workflow via bridge_workflow_event.
        assert conv_row.last_stream_seq == 1

        session_row = (await session.execute(
            select(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == conv_row.internal_session_id)
        )).scalars().first()
        assert session_row is not None
        assert session_row.user_id == "u1"

        run_row = (await session.execute(
            select(MafRoundtableRunRow).where(MafRoundtableRunRow.id == conv_row.active_run_id)
        )).scalars().first()
        assert run_row is not None
        assert run_row.session_id == conv_row.internal_session_id


@pytest.mark.asyncio
async def test_send_message_wrong_owner_404(svc):
    conv = await svc.create_conversation(user_id="u1", title="X")
    with pytest.raises(ConversationNotFoundError):
        await svc.send_message(conversation_id=conv["id"], user_id="u2", content="hack")


@pytest.mark.asyncio
async def test_concurrent_send_only_one_succeeds(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="Race")

    async def try_send(content: str):
        try:
            r = await svc.send_message(conversation_id=conv["id"], user_id="u1", content=content)
            return ("ok", r)
        except ConversationConflictError:
            return ("conflict", None)

    results = await asyncio.gather(try_send("msg1"), try_send("msg2"))
    statuses = [r[0] for r in results]
    assert statuses.count("ok") == 1
    assert statuses.count("conflict") == 1

    async with store.session_factory() as session:
        runs = (await session.execute(
            select(MafRoundtableRunRow).join(
                MafRoundtableSessionRow,
                MafRoundtableRunRow.session_id == MafRoundtableSessionRow.id,
            ).where(MafRoundtableSessionRow.user_id == "u1")
        )).scalars().all()
        assert len(runs) == 1


@pytest.mark.asyncio
async def test_message_public_dto_no_run_id(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="PubDTO")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="test content")

    messages, last_seq = await svc.list_messages(conversation_id=conv["id"], user_id="u1")
    assert len(messages) == 1
    msg = messages[0]
    assert "run_id" not in msg
    assert msg["content"] == "test content"
    assert msg["message_kind"] == "user"
    assert last_seq == 1


@pytest.mark.asyncio
async def test_last_stream_seq_contiguous(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="SeqTest")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")

    async with store.session_factory() as session:
        conv_row = (await session.execute(
            select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
        )).scalars().first()
        conv_row.safe_status = "idle"
        conv_row.active_run_id = None
        await session.commit()

    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="second")

    async with store.session_factory() as session:
        conv_row = (await session.execute(
            select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
        )).scalars().first()
        assert conv_row.last_stream_seq == 2

        msgs = (await session.execute(
            select(MafRoundtableMessageRow).where(
                MafRoundtableMessageRow.conversation_id == conv["id"]
            ).order_by(MafRoundtableMessageRow.seq)
        )).scalars().all()
        seqs = [m.seq for m in msgs]
        assert seqs == [1, 2]


@pytest.mark.asyncio
async def test_schema_rejects_forbidden_fields():
    with pytest.raises(Exception):
        CreateConversationRequest.model_validate({"title": "x", "owner_user_id": "hacker"})

    with pytest.raises(Exception):
        CreateConversationRequest.model_validate({"title": "x", "internal_session_id": "s1"})

    with pytest.raises(Exception):
        SendMessageRequest.model_validate({"content": "hi", "run_id": "r1"})

    with pytest.raises(Exception):
        SendMessageRequest.model_validate({"content": "hi", "thread_id": "t1"})


@pytest.mark.asyncio
async def test_schema_rejects_empty_and_overlong_content():
    with pytest.raises(Exception):
        SendMessageRequest.model_validate({"content": ""})

    with pytest.raises(Exception):
        SendMessageRequest.model_validate({"content": "   "})

    with pytest.raises(Exception):
        SendMessageRequest.model_validate({"content": "x" * 20001})


@pytest.mark.asyncio
async def test_task_context_whitelist(svc):
    conv = await svc.create_conversation(
        user_id="u1", title="Ctx",
        task_context={"business_code": "6BF", "secret_key": "LEAKED", "task_name": "T1"},
    )
    assert conv["task_context"] == {"business_code": "6BF", "task_name": "T1"}
    assert "secret_key" not in (conv["task_context"] or {})


@pytest.mark.asyncio
async def test_task_context_rejects_unknown_or_nested_values():
    with pytest.raises(Exception):
        CreateConversationRequest.model_validate({"task_context": {"secret_key": "nope"}})
    with pytest.raises(Exception):
        CreateConversationRequest.model_validate({"task_context": {"task_id": {"secret": "nope"}}})
    with pytest.raises(Exception):
        CreateConversationRequest.model_validate({"title": "x", "session_id": "forged"})


@pytest.mark.asyncio
async def test_disabled_feature_rejects_conversation_writes(store):
    disabled = MafConversationService(store, maf_config=SimpleNamespace(enabled=False))
    with pytest.raises(ConversationDisabledError):
        await disabled.create_conversation(user_id="u1", title="disabled")


def test_router_uses_authenticated_user_object_for_owner_scope():
    user = SimpleNamespace(id="real-user", system_role="admin")
    request = SimpleNamespace(state=SimpleNamespace(user=user))
    assert _user_ctx(request) == ("real-user", True)


@pytest.mark.asyncio
async def test_admin_cross_user_read_respects_diagnostic_config(store):
    owner_service = MafConversationService(store, maf_config=FakeMafConfig())
    conv = await owner_service.create_conversation(user_id="u1", title="Private")
    no_diagnostics = MafConversationService(
        store,
        maf_config=SimpleNamespace(enabled=True, admin_diagnostic_access=False),
    )
    with pytest.raises(ConversationNotFoundError):
        await no_diagnostics.get_conversation(
            conversation_id=conv["id"], user_id="admin", is_admin=True
        )


@pytest.mark.asyncio
async def test_conversation_internal_links_have_real_foreign_keys(store):
    async with store.session_factory() as session:
        def _fks(sync_session):
            inspector = inspect(sync_session.connection())
            return {
                (tuple(fk["constrained_columns"]), fk["referred_table"])
                for fk in inspector.get_foreign_keys("maf_roundtable_conversations")
            }

        foreign_keys = await session.run_sync(_fks)
    assert (("active_run_id",), "maf_roundtable_runs") in foreign_keys
    assert (("internal_session_id",), "maf_roundtable_sessions") in foreign_keys


@pytest.mark.asyncio
async def test_event_only_item_advances_the_same_sse_cursor(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="Unified cursor")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")

    async with store.session_factory() as session:
        async with session.begin():
            row = (
                await session.execute(
                    select(MafRoundtableConversationRow)
                    .where(MafRoundtableConversationRow.id == conv["id"])
                    .with_for_update()
                )
            ).scalars().one()
            item = await store.append_conversation_item(
                session,
                conversation=row,
                event_type="manager.intent",
                event_data={"status": "understood"},
            )
    assert item["seq"] == 2

    messages, last_stream_seq = await svc.list_messages(conversation_id=conv["id"], user_id="u1")
    assert [message["seq"] for message in messages] == [1]
    assert last_stream_seq == 2
    events = await store.conversation_events.list_events(conv["id"])
    assert [event["seq"] for event in events] == [1, 2]


@pytest.mark.asyncio
async def test_standalone_repositories_cannot_create_a_second_cursor(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="No split cursor")
    with pytest.raises(RuntimeError, match="append_conversation_item"):
        await store.messages.append_message(
            conversation_id=conv["id"], message_kind="manager", role="assistant", content="unsafe"
        )
    with pytest.raises(RuntimeError, match="append_conversation_item"):
        await store.conversation_events.append_event(
            conversation_id=conv["id"], event_type="manager.intent"
        )


@pytest.mark.asyncio
async def test_atomic_send_rolls_back_on_item_write_failure(svc, store, monkeypatch):
    conv = await svc.create_conversation(user_id="u1", title="Rollback")

    async def _fail_append(*args, **kwargs):
        raise RuntimeError("simulated item write failure")

    monkeypatch.setattr(store, "append_conversation_item", _fail_append)
    with pytest.raises(RuntimeError, match="simulated item write failure"):
        await svc.send_message(conversation_id=conv["id"], user_id="u1", content="must roll back")

    async with store.session_factory() as session:
        refreshed = (
            await session.execute(
                select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
            )
        ).scalars().one()
        assert refreshed.safe_status == "idle"
        assert refreshed.internal_session_id is None
        assert refreshed.active_run_id is None
        assert refreshed.last_stream_seq == 0
        assert (await session.execute(select(MafRoundtableMessageRow))).scalars().all() == []
        assert (await session.execute(select(MafRoundtableRunRow))).scalars().all() == []