"""Acceptance tests for the structured cancel cascade (Task 4).

Verifies that when DeerFlow cancel fails, the parent run is NOT marked
``cancelled`` —it stays non-terminal (``cancel_failed``) so the failure is
visible and retryable. A failed cancel must never masquerade as cancelled.
"""

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.deerflow_agent_adapter import DeerFlowAgentAdapter
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()


@pytest.fixture
def svc(store):
    from maf_test_helpers import make_maf_service

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


def _mock_client(*, cancel_succeeds: bool) -> MagicMock:
    client = MagicMock()
    client._app.state.run_manager.cancel = AsyncMock(return_value=None)
    client.cancel_run = AsyncMock(return_value=cancel_succeeds)
    return client


@pytest.mark.asyncio
async def test_cancel_all_succeeds_marks_cancelled(svc, store):
    """When every sub-run cancels, all_cancelled is True."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    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"])
    await svc.store.seat_runs.mark_running(active[0]["id"], deerflow_run_id="df-1")

    adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=True), service=svc, user_id="u1")
    result = await adapter.cancel(run["id"])
    assert result["all_cancelled"] is True
    assert result["failed_run_ids"] == []
    assert result["cancelled_run_ids"] == [active[0]["id"]]


@pytest.mark.asyncio
async def test_cancel_failure_keeps_run_nonterminal(svc, store):
    """When DeerFlow cancel fails, the parent run must NOT be marked cancelled.

    The adapter returns all_cancelled=False; the router would then mark the
    run ``cancel_failed`` (non-terminal). This test verifies the adapter-level
    signal that drives that decision.
    """
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    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"])
    await svc.store.seat_runs.mark_running(active[0]["id"], deerflow_run_id="df-2")

    adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=False), service=svc, user_id="u1")
    result = await adapter.cancel(run["id"])

    assert result["all_cancelled"] is False
    assert result["failed_run_ids"] == [active[0]["id"]]
    assert result["cancelled_run_ids"] == []


@pytest.mark.asyncio
async def test_mark_cancel_failed_service_method(svc, store):
    """The service ``mark_cancel_failed`` transitions a cancelling run to
    ``cancel_failed`` (non-terminal), recording a sanitized reason."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    # Put the run into cancelling via start_cancel.
    await svc.start_cancel(run_id=run["id"], user_id="u1")

    updated = await svc.mark_cancel_failed(
        run_id=run["id"], user_id="u1", reason="1 sub-run failed to cancel"
    )
    assert updated["status"] == "cancel_failed"
    assert updated["error_code"] == "CancelCascadeIncomplete"
    assert "failed to cancel" in updated["error_message_safe"]


@pytest.mark.asyncio
async def test_cancel_no_active_runs_is_all_cancelled(svc, store):
    """With no active seat runs, cancel is trivially all_cancelled (nothing to do)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=True), service=svc, user_id="u1")
    result = await adapter.cancel(run["id"])
    assert result["all_cancelled"] is True
    assert result["cancelled_run_ids"] == []