"""Tests for the MAF DeerFlow Agent Adapter (Phase 2 rework).

Tests: event normalization, context building, sanitization, cancellation,
audit event reliability, seat run persistence, and user isolation.
Uses in-memory SQLite + mocks for the DeerFlow runtime (no real model calls).
"""

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

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

from app.maf_roundtable.context_builder import build_manager_synthesis_input, build_seat_input
from app.maf_roundtable.deerflow_agent_adapter import (
    AuditEventError,
    DeerFlowAgentAdapter,
    SeatResponse,
)
from app.maf_roundtable.event_normalizer import (
    make_run_completed,
    make_turn_completed,
    make_turn_started,
    normalize_stream_event,
)
from app.maf_roundtable.runtime_client import DeerFlowRuntimeClient, ExecutionPrincipal
from app.maf_roundtable.sanitize import sanitize, sanitize_payload
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import Base


@pytest_asyncio.fixture
async def store():
    engine = create_async_engine("sqlite+aiosqlite://", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    s = make_maf_roundtable_store(sf)
    assert s is not None
    yield s
    await engine.dispose()


def _svc(store, enabled: bool = True):
    from maf_test_helpers import make_maf_service

    return make_maf_service(store, enabled=enabled)


def _mock_client() -> DeerFlowRuntimeClient:
    app = MagicMock()
    app.state.checkpointer = AsyncMock()
    app.state.thread_store = AsyncMock()
    app.state.stream_bridge = None
    app.state.run_manager = AsyncMock()
    return DeerFlowRuntimeClient(app)


# ── event_normalizer tests ────────────────────────────────────────────────────


class TestEventNormalizer:
    def test_metadata_event(self):
        events = normalize_stream_event("metadata", {"run_id": "r1", "thread_id": "t1"}, agent_id="a1", turn_index=0)
        assert len(events) == 1
        assert events[0]["event_type"] == "run.started"

    def test_error_event(self):
        events = normalize_stream_event("error", {"message": "boom", "name": "TimeoutError"}, agent_id="a1", turn_index=1)
        assert len(events) == 1
        assert events[0]["event_type"] == "run.failed"

    def test_messages_ai_with_tool_calls(self):
        data = ["msg-1", {"type": "ai", "content": "", "tool_calls": [{"name": "search", "id": "tc1", "args": {"q": "test"}}]}]
        events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
        assert any(e["event_type"] == "tool.requested" for e in events)

    def test_messages_tool_result(self):
        data = ["msg-1", {"type": "tool", "name": "search", "tool_call_id": "tc1", "content": "results...", "status": "completed"}]
        events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
        assert events[0]["event_type"] == "tool.completed"

    def test_no_skill_guessing_from_text(self):
        data = {"agent_node": {"messages": [{"type": "ai", "content": "I will use the skill to analyze this"}]}}
        events = normalize_stream_event("updates", data, agent_id="a1", turn_index=1)
        assert not any("skill" in e["event_type"] for e in events)

    def test_source_event_id_for_dedup(self):
        data = ["msg-1", {"type": "tool", "name": "search", "tool_call_id": "tc1", "content": "ok", "status": "completed"}]
        events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
        assert events[0].get("source_event_id") == "tool:tc1"

    def test_unknown_event_ignored(self):
        events = normalize_stream_event("checkpoints", {"foo": "bar"}, agent_id="a1", turn_index=0)
        assert events == []


# ── sanitization tests ────────────────────────────────────────────────────────


class TestSanitization:
    def test_sensitive_keys_redacted(self):
        data = {"token": "abc123", "name": "safe", "nested": {"password": "secret", "ok": "yes"}}
        result = sanitize(data)
        assert result["token"] == "***REDACTED***"
        assert result["name"] == "safe"
        assert result["nested"]["password"] == "***REDACTED***"
        assert result["nested"]["ok"] == "yes"

    def test_bearer_token_redacted(self):
        data = {"header": "Bearer eyJhbGciOiJIUzI1NiJ9.payload.sig"}
        result = sanitize(data)
        assert result["header"] == "***REDACTED***"

    def test_api_key_pattern_redacted(self):
        data = {"key": "sk-abcdefghijklmnopqrstuvwxyz123456"}
        result = sanitize(data)
        assert result["key"] == "***REDACTED***"

    def test_normal_values_untouched(self):
        data = {"message": "hello world", "count": 42, "items": ["a", "b"]}
        result = sanitize(data)
        assert result == data

    def test_sanitize_payload_none(self):
        assert sanitize_payload(None) is None

    def test_event_payload_sanitized(self):
        events = normalize_stream_event(
            "messages",
            ["m1", {"type": "ai", "content": "", "tool_calls": [{"name": "api_call", "id": "tc1", "args": {"api_key": "secret123", "url": "http://x"}}]}],
            agent_id="a1",
            turn_index=1,
        )
        tool_ev = next(e for e in events if e["event_type"] == "tool.requested")
        assert tool_ev["payload"]["args"]["api_key"] == "***REDACTED***"
        assert tool_ev["payload"]["args"]["url"] == "http://x"


# ── context_builder tests ─────────────────────────────────────────────────────


class TestContextBuilder:
    def test_strict_truncation(self):
        result = build_seat_input(
            task_text="x" * 20000,
            stage_instruction="",
            role_description="",
            prior_turns=[],
            max_context_chars=100,
        )
        assert len(result) <= 100

    def test_no_overflow_after_truncation(self):
        result = build_seat_input(
            task_text="task",
            stage_instruction="stage",
            role_description="role",
            prior_turns=[{"agent_id": "a", "turn_index": 1, "summary": "y" * 5000}],
            fact_ledger={"confirmed_facts": ["f" * 5000]},
            max_context_chars=200,
        )
        assert len(result) <= 200

    def test_fact_ledger_sanitized(self):
        ledger = {"confirmed_facts": ["server uses token=abc123"]}
        result = build_seat_input(
            task_text="check",
            stage_instruction="",
            role_description="",
            prior_turns=[],
            fact_ledger=ledger,
        )
        assert "abc123" not in result or "token" not in result.lower()

    def test_manager_synthesis(self):
        result = build_manager_synthesis_input(
            task_text="evaluate",
            seat_results=[{"agent_id": "risk", "content": "high risk"}],
        )
        assert "risk" in result
        assert "建议动作" in result


# ── ExecutionPrincipal tests ──────────────────────────────────────────────────


class TestExecutionPrincipal:
    def test_requires_user_id(self):
        with pytest.raises(ValueError):
            ExecutionPrincipal(user_id="", session_id="s1", run_id="r1")

    def test_valid_principal(self):
        p = ExecutionPrincipal(user_id="u1", session_id="s1", run_id="r1")
        assert p.user_id == "u1"

    def test_satisfies_current_user_protocol(self):
        from app.maf_roundtable.runtime_client import _PrincipalUser
        from deerflow.runtime.user_context import CurrentUser

        p = ExecutionPrincipal(user_id="test-user", session_id="s", run_id="r")
        user = _PrincipalUser(p)
        assert isinstance(user, CurrentUser)
        assert user.id == "test-user"


# ── adapter unit tests ────────────────────────────────────────────────────────


class TestDeerFlowAgentAdapter:
    @pytest.mark.asyncio
    async def test_ensure_threads_creates_and_binds(self, store):
        svc = _svc(store)
        session = await svc.create_session(
            user_id="u1",
            body={"task_text": "test", "participants": [{"agent_id": "a1"}, {"agent_id": "a2"}]},
        )
        run = await svc.create_run(session_id=session["id"], user_id="u1")
        client = _mock_client()
        client._app.state.checkpointer.aput = AsyncMock()
        client._app.state.thread_store.create = AsyncMock()

        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
        thread_map = await adapter.ensure_threads(session["id"], run["id"], session["participants"])

        assert len(thread_map) == 2
        assert thread_map["a1"] != thread_map["a2"]

        updated = await svc.get_session(session_id=session["id"], user_id="u1")
        for p in updated["participants"]:
            assert p["thread_id"] is not None

    @pytest.mark.asyncio
    async def test_ensure_threads_reuses_existing(self, store):
        svc = _svc(store)
        session = await svc.create_session(
            user_id="u1",
            body={"task_text": "test", "participants": [{"agent_id": "a1", "thread_id": "existing-t1"}]},
        )
        run = await svc.create_run(session_id=session["id"], user_id="u1")
        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
        thread_map = await adapter.ensure_threads(session["id"], run["id"], session["participants"])
        assert thread_map["a1"] == "existing-t1"

    @pytest.mark.asyncio
    async def test_cancel_cascades_via_seat_runs(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
        active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
        seat_run_id = active[0]["id"]
        await svc.store.seat_runs.mark_running(seat_run_id, deerflow_run_id="df-run-1")

        client = _mock_client()
        client._app.state.run_manager.cancel = AsyncMock()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
        result = await adapter.cancel(run["id"])

        # Structured cancel result (Task 4): all_cancelled when every active
        # sub-run was confirmed cancelled.
        assert result["all_cancelled"] is True
        assert result["cancelled_run_ids"] == [seat_run_id]
        assert result["failed_run_ids"] == []
        client._app.state.run_manager.cancel.assert_called_once_with("df-run-1", action="interrupt")
        remaining = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
        assert len(remaining) == 0

    @pytest.mark.asyncio
    async def test_critical_event_failure_raises(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        original_append = svc.append_event
        async def _failing_append(**kwargs):
            raise RuntimeError("DB connection lost")
        svc.append_event = _failing_append

        with pytest.raises(AuditEventError):
            await adapter._emit_critical(run["id"], make_turn_started("a1", 0))

        svc.append_event = original_append

    @pytest.mark.asyncio
    async def test_noncritical_event_degrades_gracefully(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        async def _failing_append(**kwargs):
            raise RuntimeError("DB connection lost")
        svc.append_event = _failing_append

        await adapter._emit(run["id"], {"event_type": "agent.message.delta", "agent_id": "a1", "turn_index": 0, "payload": {"text": "hi"}})

    @pytest.mark.asyncio
    async def test_seat_response_dataclass(self):
        r = SeatResponse(agent_id="a1", turn_index=1, content="hello", tool_calls=[{"name": "search"}])
        assert r.agent_id == "a1"
        assert r.error is None

    @pytest.mark.asyncio
    async def test_latest_ai_text_extraction(self, store):
        svc = _svc(store)
        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
        messages = [
            {"type": "human", "content": "hello"},
            {"type": "ai", "content": "", "tool_calls": [{"name": "search", "args": {}, "id": "1"}]},
            {"type": "ai", "content": "The answer is 42."},
        ]
        assert adapter._latest_ai_text(messages) == "The answer is 42."

    @pytest.mark.asyncio
    async def test_tool_calls_since_last_human(self, store):
        svc = _svc(store)
        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
        messages = [
            {"type": "human", "content": "old"},
            {"type": "ai", "content": "", "tool_calls": [{"name": "old_tool", "args": {}, "id": "0"}]},
            {"type": "human", "content": "new"},
            {"type": "ai", "content": "", "tool_calls": [{"name": "search", "args": {"q": "x"}, "id": "1"}]},
        ]
        calls = adapter._tool_calls_since_last_human(messages)
        assert len(calls) == 1
        assert calls[0]["name"] == "search"


# ── seat_runs persistence tests ───────────────────────────────────────────────


class TestSeatRunPersistence:
    @pytest.mark.asyncio
    async def test_create_and_lifecycle(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        sr = await svc.store.seat_runs.create_seat_run(
            maf_run_id=run["id"], agent_id="expert-a", turn_index=1, deerflow_thread_id="t1"
        )
        assert sr["status"] == "pending"

        await svc.store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")
        active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
        assert len(active) == 1
        assert active[0]["deerflow_run_id"] == "df-1"

        await svc.store.seat_runs.mark_completed(sr["id"], checkpoint_id="cp-1")
        active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
        assert len(active) == 0

    @pytest.mark.asyncio
    async def test_cancel_all_active(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        sr1 = await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
        sr2 = await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a2", turn_index=1)
        await svc.store.seat_runs.mark_running(sr1["id"], deerflow_run_id="df-1")
        await svc.store.seat_runs.mark_running(sr2["id"], deerflow_run_id="df-2")

        cancelled = await svc.store.seat_runs.cancel_all_active(run["id"])
        # Two-phase cancel: returns {seat_run_id, deerflow_run_id} dicts and
        # leaves rows in 'cancelling' (caller finalizes to cancelled).
        assert {c["deerflow_run_id"] for c in cancelled} == {"df-1", "df-2"}
        assert {c["seat_run_id"] for c in cancelled} == {sr1["id"], sr2["id"]}
        # 'cancelling' is not an active status, so active count drops to 0.
        active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
        assert len(active) == 0
        # Rows are mid-cancel, not yet finalized.
        all_runs = await svc.store.seat_runs.get_by_maf_run(run["id"])
        assert {r["status"] for r in all_runs} == {"cancelling"}

    @pytest.mark.asyncio
    async def test_unique_constraint(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
        with pytest.raises(Exception):
            await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)


# ── event flow integration tests ──────────────────────────────────────────────


class TestAdapterEventFlow:
    @pytest.mark.asyncio
    async def test_full_event_sequence_persisted(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "analyze"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        await adapter._emit_critical(run["id"], make_turn_started("expert", 1))
        await adapter._emit(run["id"], {"event_type": "tool.requested", "agent_id": "expert", "turn_index": 1, "payload": {"tool_name": "search"}})
        await adapter._emit(run["id"], {"event_type": "tool.completed", "agent_id": "expert", "turn_index": 1, "payload": {"tool_name": "search", "status": "succeeded"}})
        await adapter._emit_critical(run["id"], make_turn_completed("expert", 1, content="done"))
        await adapter._emit_critical(run["id"], make_run_completed("expert", 1))

        events, latest = await svc.list_events(run_id=run["id"], user_id="u1")
        assert len(events) == 5
        assert latest == 5
        types = [e["event_type"] for e in events]
        assert types == ["agent.turn_started", "tool.requested", "tool.completed", "agent.message.completed", "run.completed"]

    @pytest.mark.asyncio
    async def test_dedup_via_source_event_id(self, store):
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        ev = {"event_type": "tool.completed", "agent_id": "a1", "turn_index": 1, "payload": {"tool_name": "x"}, "source_event_id": "tool:tc1"}
        await adapter._emit(run["id"], ev)
        await adapter._emit(run["id"], ev)

        events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
        assert len(events) == 1

    @pytest.mark.asyncio
    async def test_emit_critical_passes_persisted_audit_id_to_bridge(self, store):
        """_emit_critical must capture the persisted audit row id and pass it
        as source_audit_event_id to _bridge_conversation_event, so projection
        dedup is keyed on the real audit row — not an ephemeral input field."""
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        bridge_calls: list[dict] = []
        original_bridge = adapter._bridge_conversation_event

        async def capture_bridge(run_id, event, *, source_audit_event_id=None):
            bridge_calls.append({
                "run_id": run_id,
                "event_type": event.get("event_type"),
                "source_audit_event_id": source_audit_event_id,
            })
            await original_bridge(run_id, event, source_audit_event_id=source_audit_event_id)

        adapter._bridge_conversation_event = capture_bridge

        critical_event = make_turn_started("expert", 1)
        await adapter._emit_critical(run["id"], critical_event)

        assert len(bridge_calls) == 1
        assert bridge_calls[0]["source_audit_event_id"] is not None
        assert isinstance(bridge_calls[0]["source_audit_event_id"], str)
        assert len(bridge_calls[0]["source_audit_event_id"]) > 0

    @pytest.mark.asyncio
    async def test_emit_passes_persisted_audit_id_for_non_critical(self, store):
        """_emit (non-critical path) must also pass persisted audit row id."""
        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        bridge_calls: list[dict] = []
        original_bridge = adapter._bridge_conversation_event

        async def capture_bridge(run_id, event, *, source_audit_event_id=None):
            bridge_calls.append({
                "source_audit_event_id": source_audit_event_id,
            })
            await original_bridge(run_id, event, source_audit_event_id=source_audit_event_id)

        adapter._bridge_conversation_event = capture_bridge

        ev = {
            "event_type": "agent.message.delta",
            "agent_id": "a1",
            "turn_index": 0,
            "payload": {"text": "hello"},
            "source_event_id": "stream:evt_001",
        }
        await adapter._emit(run["id"], ev)

        assert len(bridge_calls) == 1
        assert bridge_calls[0]["source_audit_event_id"] is not None

    @pytest.mark.asyncio
    async def test_emit_critical_dedup_across_simulated_restart(self, store):
        """Simulate: emit critical event, then 'restart' and re-emit the same
        event via reconcile. The conversation projection must be deduped via
        the persisted audit row id — no duplicate conversation events."""
        from app.maf_roundtable.conversation_event_hub import ConversationEventHub
        from app.maf_roundtable.conversation_service import MafConversationService

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

        hub = ConversationEventHub(heartbeat_interval=0.3)
        conv_svc = MafConversationService(store, maf_config=FakeMafConfig(), event_hub=hub)

        conv = await conv_svc.create_conversation(user_id="u1", title="RestartDedup")

        from sqlalchemy import select

        from deerflow.persistence.maf_roundtable.model import (
            MafRoundtableConversationRow,
            MafRoundtableRunRow,
            MafRoundtableSessionRow,
        )

        svc = _svc(store)
        session = await svc.create_session(user_id="u1", body={"task_text": "test"})
        run = await svc.create_run(session_id=session["id"], user_id="u1")

        async with store.session_factory() as db:
            async with db.begin():
                conv_row = (
                    await db.execute(
                        select(MafRoundtableConversationRow).where(
                            MafRoundtableConversationRow.id == conv["id"]
                        ).with_for_update()
                    )
                ).scalars().first()
                assert conv_row is not None
                sess_row = (
                    await db.execute(
                        select(MafRoundtableSessionRow).where(
                            MafRoundtableSessionRow.id == session["id"]
                        )
                    )
                ).scalars().first()
                conv_row.internal_session_id = sess_row.id
                run_row = (
                    await db.execute(
                        select(MafRoundtableRunRow).where(
                            MafRoundtableRunRow.id == run["id"]
                        )
                    )
                ).scalars().first()
                assert run_row is not None

        client = _mock_client()
        adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")

        critical_event = {
            "event_type": "action.waiting_approval",
            "agent_id": "seat_a",
            "turn_index": 1,
            "payload": {"tool_name": "write_file", "summary": "high risk"},
        }
        await adapter._emit_critical(run["id"], critical_event)

        await conv_svc.reconcile_conversation_events(run_id=run["id"])

        events = await store.conversation_events.list_events(conv["id"], after_seq=0)
        approval_events = [e for e in events if e["type"] == "approval.required"]
        assert len(approval_events) == 1

        hub.shutdown()