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

import pytest
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.routers import parallel_agent_tasks
from deerflow.persistence.base import Base
from deerflow.persistence.parallel_agent_tasks import ParallelAgentTaskRepository


@pytest.mark.asyncio
async def test_parallel_task_repository_rejects_stale_revision(tmp_path) -> None:
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'strategy-project-cas.db'}")
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    store = ParallelAgentTaskRepository(async_sessionmaker(engine, expire_on_commit=False))
    try:
        original = await store.create_task(
            "user-1",
            {
                "id": "strategy-selection-cas",
                "title": "original",
                "messages": [{"snapshot": "original"}],
            },
        )
        await asyncio.sleep(0.002)
        newer = await store.update_task(
            "strategy-selection-cas",
            "user-1",
            title="newer",
            messages=[{"snapshot": "newer"}],
            expected_updated_at=original["updated_at"],
        )
        assert newer is not None

        stale = await store.update_task(
            "strategy-selection-cas",
            "user-1",
            title="stale",
            messages=[{"snapshot": "stale"}],
            expected_updated_at=original["updated_at"],
        )
        assert stale is None
        final = await store.get_task("strategy-selection-cas", "user-1")
        assert final is not None
        assert final["title"] == "newer"
        assert final["messages"] == [{"snapshot": "newer"}]

        lightweight = await store.list_tasks("user-1")
        assert "messages" not in lightweight[0]
        with_snapshots = await store.list_tasks("user-1", include_messages=True)
        assert with_snapshots[0]["messages"] == [{"snapshot": "newer"}]

        await store.create_task(
            "user-1",
            {
                "id": "parallel-unrelated",
                "title": "unrelated",
                "messages": [{"large": "transcript"}],
            },
        )
        await asyncio.sleep(0.002)
        await store.create_task(
            "user-1",
            {
                "id": "strategy-selection-second",
                "title": "second",
                "messages": [{"snapshot": "second"}],
            },
        )
        first_page = await store.list_tasks(
            "user-1",
            limit=1,
            id_prefix="strategy-selection-",
            include_messages=True,
        )
        second_page = await store.list_tasks(
            "user-1",
            limit=1,
            offset=1,
            id_prefix="strategy-selection-",
            include_messages=True,
        )
        assert [first_page[0]["id"], second_page[0]["id"]] == [
            "strategy-selection-second",
            "strategy-selection-cas",
        ]
        assert first_page[0]["messages"] == [{"snapshot": "second"}]
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_parallel_task_route_reports_revision_conflict() -> None:
    class ConflictStore:
        async def update_task(self, *_args, **_kwargs):
            return None

        async def get_task(self, *_args, **_kwargs):
            return {"id": "strategy-selection-cas"}

    request = SimpleNamespace(
        state=SimpleNamespace(user=SimpleNamespace(id="user-1")),
        app=SimpleNamespace(state=SimpleNamespace(parallel_agent_task_store=ConflictStore())),
    )
    body = parallel_agent_tasks.TaskUpdateRequest(
        title="stale",
        expected_updated_at=datetime.now(UTC),
    )

    with pytest.raises(HTTPException) as error:
        await parallel_agent_tasks.update_task(
            request,
            "strategy-selection-cas",
            body,
        )

    assert error.value.status_code == 409
    assert "其他页面更新" in str(error.value.detail)