"""P1 concurrency + single-transaction launch tests for MAF roundtable.

Covers:
- **Concurrent run launch**: ``asyncio.gather`` of ≥ ``create_run`` calls must
  yield exactly one active run; the rest get ``MafConflictError``.
- **Single-transaction atomicity**: flush / pre-commit failures roll back fully —
  session stays at its prior status, no orphan run, no residual ``starting``.
- **Success shape**: session ``running``, unique run ``pending``.
- **Cancel then re-launch** works.
- **Concurrent approval decision**: parallel approve + reject →exactly one winner.
"""

from __future__ import annotations

import asyncio
from unittest.mock import patch

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

from app.maf_roundtable.service import (
    MafConflictError,
)
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store


@pytest_asyncio.fixture
async def store(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_conc.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()


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

    return make_maf_service(store, enabled=enabled, admin_diagnostic_access=admin_diagnostic_access)


# ── P1: concurrent run launch (single-tx CAS) ─────────────────────────────


@pytest.mark.asyncio
async def test_concurrent_create_run_only_one_wins(store):
    """8 parallel ``create_run`` calls →exactly one active run, 7 conflicts."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})

    async def _attempt() -> tuple[str, object]:
        try:
            run = await svc.create_run(session_id=s["id"], user_id="u1")
            return ("ok", run)
        except MafConflictError as exc:
            return ("conflict", exc)

    results = await asyncio.gather(*[_attempt() for _ in range(8)])
    oks = [r for r in results if r[0] == "ok"]
    conflicts = [r for r in results if r[0] == "conflict"]

    assert len(oks) == 1, f"expected exactly one winner, got {len(oks)}"
    assert len(conflicts) == 7

    runs = await svc.list_runs(session_id=s["id"], user_id="u1")
    active = [r for r in runs if r["status"] in ("pending", "running", "waiting_approval", "waiting_input")]
    assert len(active) == 1
    assert active[0]["id"] == oks[0][1]["id"]
    assert active[0]["status"] == "pending"

    refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed["status"] == "running"


@pytest.mark.asyncio
async def test_create_run_success_session_running_run_pending(store):
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    assert run["status"] == "pending"
    refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed["status"] == "running"
    assert len(await svc.list_runs(session_id=s["id"], user_id="u1")) == 1


@pytest.mark.asyncio
async def test_concurrent_create_run_second_launch_after_cancel(store):
    """After the winning run is cancelled, a fresh launch succeeds."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})

    results = await asyncio.gather(
        *[svc.create_run(session_id=s["id"], user_id="u1") for _ in range(3)],
        return_exceptions=True,
    )
    winners = [r for r in results if not isinstance(r, Exception)]
    assert len(winners) == 1

    await svc.cancel_run(run_id=winners[0]["id"], user_id="u1")
    run2 = await svc.create_run(session_id=s["id"], user_id="u1")
    assert run2["status"] == "pending"
    refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed["status"] == "running"


@pytest.mark.asyncio
async def test_flush_failure_rolls_back_session_and_no_run(store):
    """If flush() fails after the claim UPDATE, the whole transaction rolls back:
    session stays at its prior status and no run row is persisted.
    """
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    prior_status = s["status"]
    assert prior_status == "draft"

    async def boom_flush(self, *args, **kwargs):
        raise RuntimeError("simulated flush failure")

    with patch.object(AsyncSession, "flush", boom_flush):
        with pytest.raises(RuntimeError, match="simulated flush failure"):
            await svc.create_run(session_id=s["id"], user_id="u1")

    refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed["status"] == prior_status
    assert refreshed["status"] != "starting"
    assert await svc.list_runs(session_id=s["id"], user_id="u1") == []

    # A subsequent healthy launch still works.
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    assert run["status"] == "pending"


@pytest.mark.asyncio
async def test_pre_commit_failure_leaves_no_starting_residue(store):
    """If an error occurs after flush but before commit, the transaction rolls
    back: session must not residual ``starting`` / ``running``, and no run row.
    """
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})

    real_flush = AsyncSession.flush

    async def flush_then_boom(self, *args, **kwargs):
        await real_flush(self, *args, **kwargs)
        raise RuntimeError("simulated post-flush pre-commit failure")

    with patch.object(AsyncSession, "flush", flush_then_boom):
        with pytest.raises(RuntimeError, match="simulated post-flush pre-commit failure"):
            await svc.create_run(session_id=s["id"], user_id="u1")

    refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed["status"] == "draft"
    assert refreshed["status"] != "starting"
    assert await svc.list_runs(session_id=s["id"], user_id="u1") == []


# ── P1: concurrent approval decision ──────────────────────────────────────


@pytest.mark.asyncio
async def test_concurrent_approval_decision_only_one_wins(store):
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    appr = await store.approvals.create_approval(
        {"session_id": s["id"], "run_id": run["id"], "summary": "高风险写入", "status": "pending"}
    )

    async def _decide(decision: str, who: str) -> tuple[str, object]:
        try:
            decided = await svc.decide_approval(approval_id=appr["id"], decision=decision, user_id=who)
            return ("ok", decided)
        except MafConflictError as exc:
            return ("conflict", exc)

    results = await asyncio.gather(
        *[_decide("approved", "u1") for _ in range(4)],
        *[_decide("rejected", "u1") for _ in range(4)],
    )
    oks = [r for r in results if r[0] == "ok"]
    conflicts = [r for r in results if r[0] == "conflict"]

    assert len(oks) == 1, f"expected exactly one winning decision, got {len(oks)}"
    assert len(conflicts) == 7

    winner = oks[0][1]
    final = await store.approvals.get_approval(appr["id"])
    assert final["status"] == winner["status"]
    assert final["decided_by"] == winner["decided_by"] == "u1"
    assert final["decided_at"] == winner["decided_at"]
    assert final["decided_at"] is not None


@pytest.mark.asyncio
async def test_approval_decision_cannot_be_overwritten(store):
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    appr = await store.approvals.create_approval(
        {"session_id": s["id"], "run_id": run["id"], "summary": "x", "status": "pending"}
    )
    first = await svc.decide_approval(approval_id=appr["id"], decision="approved", user_id="u1")
    with pytest.raises(MafConflictError):
        await svc.decide_approval(approval_id=appr["id"], decision="rejected", user_id="u1")
    final = await store.approvals.get_approval(appr["id"])
    assert final["status"] == first["status"] == "approved"
    assert final["decided_by"] == "u1"


# ── P1: cancel_run single-transaction atomicity ───────────────────────────


@pytest.mark.asyncio
async def test_cancel_run_flush_failure_rolls_back_run_and_session(store):
    """If the cancel transaction's flush fails, run stays pending and session
    stays running —nothing is half-committed."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    assert run["status"] == "pending"

    async def boom_flush(self, *args, **kwargs):
        raise RuntimeError("simulated cancel flush failure")

    with patch.object(AsyncSession, "flush", boom_flush):
        with pytest.raises(RuntimeError, match="simulated cancel flush failure"):
            await svc.cancel_run(run_id=run["id"], user_id="u1")

    refreshed_run = await svc.get_run(run_id=run["id"], user_id="u1")
    assert refreshed_run["status"] == "pending"
    assert refreshed_run["finished_at"] is None
    refreshed_session = await svc.get_session(session_id=s["id"], user_id="u1")
    assert refreshed_session["status"] == "running"

    # A healthy cancel afterwards succeeds.
    cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1")
    assert cancelled["status"] == "cancelled"


@pytest.mark.asyncio
async def test_cancel_repairs_historical_inconsistent_session(store):
    """Historical half-completed data: run=cancelled but session still running.
    Re-cancelling returns the cancelled run and repairs the session."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    # Manufacture the inconsistency: run cancelled, session left running.
    await store.runs.set_status(run["id"], "cancelled")
    await store.sessions.set_status(s["id"], "running")
    assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "running"

    result = await svc.cancel_run(run_id=run["id"], user_id="u1")
    assert result["status"] == "cancelled"
    assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "cancelled"


@pytest.mark.asyncio
async def test_cancel_terminal_run_does_not_change_run(store):
    """A completed/failed run is not re-cancelled; its status is returned as-is."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    await store.runs.set_status(run["id"], "completed")
    await store.sessions.set_status(s["id"], "completed")

    result = await svc.cancel_run(run_id=run["id"], user_id="u1")
    assert result["status"] == "completed"
    assert (await svc.get_run(run_id=run["id"], user_id="u1"))["status"] == "completed"


@pytest.mark.asyncio
async def test_concurrent_cancel_same_run_consistent(store):
    """Two concurrent cancels of the same run →run and session both cancelled,
    no exceptions, no inconsistent state."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    results = await asyncio.gather(
        *[svc.cancel_run(run_id=run["id"], user_id="u1") for _ in range(2)],
        return_exceptions=True,
    )
    assert not any(isinstance(r, Exception) for r in results), results
    assert all(r["status"] == "cancelled" for r in results)

    assert (await svc.get_run(run_id=run["id"], user_id="u1"))["status"] == "cancelled"
    assert (await svc.get_session(session_id=s["id"], user_id="u1"))["status"] == "cancelled"


# ── P2: pending run time semantics ────────────────────────────────────────


@pytest.mark.asyncio
async def test_pending_run_has_no_started_at(store):
    """A freshly-created pending run is queued, not started: started_at is None."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    assert run["status"] == "pending"
    assert run["started_at"] is None


@pytest.mark.asyncio
async def test_cancel_pending_run_sets_finished_at(store):
    """Cancelling a pending run sets finished_at (terminal) while started_at
    stays None (it never actually ran)."""
    svc = _svc(store)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="u1")
    cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1")
    assert cancelled["status"] == "cancelled"
    assert cancelled["finished_at"] is not None
    assert cancelled["started_at"] is None