"""Entry-level MAF Phase-5 approval and target-lease acceptance tests."""

from __future__ import annotations

import asyncio
import threading
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest
from fastapi import FastAPI, Request
from httpx import ASGITransport, AsyncClient
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langgraph.checkpoint.memory import InMemorySaver
from sqlalchemy import update
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.app import _scan_maf_approval_expiry
from app.gateway.routers.maf_roundtable import router as maf_router
from app.maf_roundtable.dispatcher import get_dispatcher
from app.maf_roundtable.write_guard_bridge import register_write_guard_service
from deerflow.config import get_app_config
from deerflow.config.maf_roundtable_config import MafRoundtableConfig, MafToolMetadata
from deerflow.config.tool_config import ToolConfig
from deerflow.config.write_guard_config import WriteGuardConfig, WriteGuardProviderConfig
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import MafRoundtableApprovalRow
from deerflow.runtime import RunManager
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge

TEST_TOOL = "maf_test_outbox_write"


class _User:
    def __init__(self, user_id: str) -> None:
        self.id = user_id
        self.system_role = "user"


class _ThreadStore:
    def __init__(self) -> None:
        self.rows: dict[str, dict] = {}

    async def create(self, thread_id, assistant_id=None, metadata=None):
        self.rows[thread_id] = {
            "thread_id": thread_id,
            "assistant_id": assistant_id,
            "metadata": metadata or {},
            "status": "idle",
        }

    async def get(self, thread_id):
        return self.rows.get(thread_id)

    async def update_status(self, thread_id, status):
        self.rows.setdefault(thread_id, {})["status"] = status

    async def update_assistant_id(self, thread_id, assistant_id):
        self.rows.setdefault(thread_id, {})["assistant_id"] = assistant_id

    async def update_metadata(self, thread_id, metadata):
        self.rows.setdefault(thread_id, {}).setdefault("metadata", {}).update(metadata)

    async def update_display_name(self, thread_id, display_name):
        self.rows.setdefault(thread_id, {})["display_name"] = display_name


class _DeterministicModel(BaseChatModel):
    """Deterministic model boundary: propose one write, then finish."""

    calls: int = 0

    @property
    def _llm_type(self) -> str:
        return "maf-phase5-deterministic"

    def bind_tools(self, tools, **kwargs):  # noqa: ARG002
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):  # noqa: ARG002
        if self.calls == 0:
            message = AIMessage(
                content="",
                tool_calls=[{
                    "name": TEST_TOOL,
                    "args": {"operation_id": "phase5-target", "value": "controlled"},
                    "id": "phase5-tool-call",
                    "type": "tool_call",
                }],
            )
        else:
            message = AIMessage(content="deterministic completion")
        self.calls += 1
        return ChatResult(generations=[ChatGeneration(message=message)])


def _app_config():
    config = get_app_config().model_copy(deep=True)
    config.tools = [
        *config.tools,
        ToolConfig(
            name=TEST_TOOL,
            group="maf-e2e",
            use="deerflow.tools.builtins.maf_test_outbox:maf_test_outbox_write",
        ),
    ]
    config.write_guard = WriteGuardConfig(
        enabled=True,
        provider=WriteGuardProviderConfig(
            policy_use="app.maf_roundtable.write_guard_bridge:make_policy",
            interceptor_use="app.maf_roundtable.write_guard_bridge:make_interceptor",
        ),
    )
    config.maf_roundtable = MafRoundtableConfig(
        enabled=True,
        default_write_policy="confirm_all_writes",
        approval_timeout_seconds=300,
        approval_scan_interval_seconds=5,
        tool_metadata={
            TEST_TOOL: MafToolMetadata(
                side_effect="write",
                risk_level="R2",
                target_key_fields=["operation_id"],
                target_key_namespace="phase5-e2e",
            )
        },
    )
    return config


class _VisibleAgentStore:
    async def get_visible(self, agent_id: str, user_id: str):  # noqa: ARG002
        return {"id": agent_id, "user_id": user_id}

    async def ensure_builtin(self, record):  # noqa: ARG002
        return None


def _install_test_app(app: FastAPI, service, config) -> None:
    app.state.config = config
    app.state.maf_roundtable_service = service
    app.state.maf_auto_dispatch = True
    app.state.checkpointer = InMemorySaver()
    app.state.thread_store = _ThreadStore()
    app.state.run_manager = RunManager()
    app.state.stream_bridge = MemoryStreamBridge()
    app.state.run_event_store = MemoryRunEventStore()
    app.state.store = None
    app.state.agent_store = _VisibleAgentStore()
    register_write_guard_service(service)
    app.include_router(maf_router)


async def _wait_dispatch(dispatcher, run_id: str) -> None:
    task = dispatcher._tasks.get(run_id)  # noqa: SLF001 - entry test waits on the real task
    assert task is not None
    await task


@pytest.fixture
def deterministic_agent(monkeypatch, tmp_path):
    agent_config = SimpleNamespace(
        tool_groups=["maf-e2e"],
        id="seat-1",
        description="MAF E2E test seat",
        model=None,
        skills=[],
        subagent_enabled=False,
        name="MAF E2E seat",
    )
    paths = (
        "deerflow.config.agents_config.load_agent_config",
        "app.maf_roundtable.capability_snapshot.load_agent_config",
        "deerflow.agents.lead_agent.agent.load_agent_config",
    )
    for path in paths:
        monkeypatch.setattr(path, lambda _agent_id, cfg=agent_config: cfg)
    monkeypatch.setattr("deerflow.agents.lead_agent.agent.load_agent_soul", lambda _agent_id: "")
    monkeypatch.setattr(
        "deerflow.agents.lead_agent.agent.create_chat_model",
        lambda **_kwargs: _DeterministicModel(),
    )
    config = _app_config()
    config.title.enabled = False
    monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "phase5-entry-test-secret")
    monkeypatch.setenv("MAF_E2E_OUTBOX_DIR", str(tmp_path))
    return config


async def _make_app(tmp_path, config):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_entry.db'}")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    store = make_maf_roundtable_store(sf)
    from app.maf_roundtable.service import MafRoundtableService

    service = MafRoundtableService(
        store,
        enabled=True,
        admin_diagnostic_access=False,
        config=config.maf_roundtable,
        agent_store=_VisibleAgentStore(),
        app_config=config,
    )
    app = FastAPI()
    _install_test_app(app, service, config)

    @app.middleware("http")
    async def test_auth(request: Request, call_next):
        request.state.user = _User("u1")
        return await call_next(request)

    return app, service, engine


async def _create_waiting_run(client: AsyncClient, dispatcher, *, task_text: str, service=None) -> tuple[str, dict]:
    session_response = await client.post(
        "/api/maf-roundtable/sessions",
        json={
            "task_text": task_text,
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "seat-1", "role": "seat"}],
        },
    )
    assert session_response.status_code == 200, session_response.text
    session = session_response.json()
    run_response = await client.post(
        f"/api/maf-roundtable/sessions/{session['id']}/runs",
        json={},
    )
    assert run_response.status_code == 200, run_response.text
    run = run_response.json()
    await _wait_dispatch(dispatcher, run["id"])
    run_state = (await client.get(f"/api/maf-roundtable/runs/{run['id']}")).json()
    if run_state["status"] != "waiting_approval" and service is not None:
        actions = await service.store.actions.list_actions(run_id=run["id"])
        raise AssertionError(f"unexpected run state: {run_state}; actions={actions}")
    assert run_state["status"] == "waiting_approval"
    return session["id"], run


@pytest.mark.asyncio
async def test_entry_http_approval_reaches_real_toolnode_once(tmp_path, deterministic_agent):
    app, service, engine = await _make_app(tmp_path, deterministic_agent)
    dispatcher = get_dispatcher(app)
    try:
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            session_id, run = await _create_waiting_run(client, dispatcher, task_text="entry approval", service=service)
            approvals = (await client.get(f"/api/maf-roundtable/runs/{run['id']}/approvals")).json()["approvals"]
            assert len(approvals) == 1
            approval_response = await client.post(
                f"/api/maf-roundtable/approvals/{approvals[0]['id']}/decision",
                json={"decision": "approved"},
            )
            assert approval_response.status_code == 200, approval_response.text
            action = (await client.get(f"/api/maf-roundtable/runs/{run['id']}/actions")).json()["actions"][0]
            assert action["status"] == "succeeded"
            outbox = tmp_path / "phase5-target.jsonl"
            assert outbox.exists()
            assert outbox.read_text(encoding="utf-8").count("\n") == 1
            assert (await client.get(f"/api/maf-roundtable/sessions/{session_id}")).json()["status"] == "completed"
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_entry_approval_scanner_expires_and_finishes_without_tool(tmp_path, deterministic_agent):
    app, service, engine = await _make_app(tmp_path, deterministic_agent)
    dispatcher = get_dispatcher(app)
    try:
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            _session_id, run = await _create_waiting_run(
                client, dispatcher, task_text="scanner expiry", service=service
            )
            approval = (await client.get(
                f"/api/maf-roundtable/runs/{run['id']}/approvals"
            )).json()["approvals"][0]
            async with service.store.session_factory() as db:
                await db.execute(
                    update(MafRoundtableApprovalRow)
                    .where(MafRoundtableApprovalRow.id == approval["id"])
                    .values(expires_at=datetime.now(UTC) - timedelta(seconds=1))
                )
                await db.commit()

            assert await _scan_maf_approval_expiry(app) == [run["id"]]
            final_run = (await client.get(f"/api/maf-roundtable/runs/{run['id']}")).json()
            assert final_run["status"] == "completed"
            action = (await client.get(
                f"/api/maf-roundtable/runs/{run['id']}/actions"
            )).json()["actions"][0]
            assert action["status"] == "rejected"
            assert not (tmp_path / "phase5-target.jsonl").exists()
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_entry_http_same_target_waits_for_explicit_resume(tmp_path, deterministic_agent):
    app, service, engine = await _make_app(tmp_path, deterministic_agent)
    dispatcher = get_dispatcher(app)
    from deerflow.tools.builtins import maf_test_outbox

    started = threading.Event()
    release = threading.Event()
    maf_test_outbox.configure_test_block(started=started, release=release)
    try:
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            _session_one, run_one = await _create_waiting_run(client, dispatcher, task_text="lease one", service=service)
            approval_one = (await client.get(f"/api/maf-roundtable/runs/{run_one['id']}/approvals")).json()["approvals"][0]
            first_decision = asyncio.create_task(client.post(
                f"/api/maf-roundtable/approvals/{approval_one['id']}/decision",
                json={"decision": "approved"},
            ))
            assert await asyncio.to_thread(started.wait, 10)

            _session_two, run_two = await _create_waiting_run(client, dispatcher, task_text="lease two", service=service)
            approval_two = (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}/approvals")).json()["approvals"][0]
            second_decision = await client.post(
                f"/api/maf-roundtable/approvals/{approval_two['id']}/decision",
                json={"decision": "approved"},
            )
            assert second_decision.status_code == 200
            assert (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}")).json()["status"] == "waiting_approval"
            second_action = (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}/actions")).json()["actions"][0]
            assert second_action["status"] == "approved"

            release.set()
            assert (await first_decision).status_code == 200
            assert (await client.get(f"/api/maf-roundtable/runs/{run_one['id']}")).json()["status"] == "completed"

            resumed = await client.post(f"/api/maf-roundtable/runs/{run_two['id']}/resume")
            assert resumed.status_code == 200, resumed.text
            assert (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}")).json()["status"] == "completed"
            outbox = tmp_path / "phase5-target.jsonl"
            assert outbox.read_text(encoding="utf-8").count("\n") == 2
    finally:
        release.set()
        maf_test_outbox.clear_test_block()
        await engine.dispose()