"""Phase-3 Round-2 acceptance tests (P0-3): real manager workflow coexistence.

Drives the REAL workflow executors (PlannerExecutor → SeatExecutor ×N →
CrossReviewExecutor → ApprovalGateExecutor → FinalizerExecutor) through a
manager + 2-seat structured run, with a fake DeerFlow adapter whose
``run_seat_turn`` records each call's ``execution_stage`` into a REAL seat-run
store. This proves the Phase-3 disambiguation invariants end-to-end:

- manager planner / cross_review / finalizer all share the SAME business round
  (turn_index) but occupy distinct seat-run rows via ``execution_stage`` — they
  must coexist without a four-tuple collision.
- an expert seat turn and its one-shot ``format_repair`` retry coexist for the
  same (agent, turn).
- invalid expert output triggers exactly one ``format_repair`` DeerFlow call,
  and both rows persist.
- invalid cross-review output triggers exactly one ``cross_review_repair``
  DeerFlow call, and ``cross_review`` + ``cross_review_repair`` rows coexist.
"""

from __future__ import annotations

import json
from datetime import datetime
from unittest.mock import AsyncMock

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

from app.maf_roundtable.deerflow_agent_adapter import SeatResponse
from app.maf_roundtable.workflow_executors import (
    ApprovalGateExecutor,
    CrossReviewExecutor,
    FinalizerExecutor,
    PlannerExecutor,
    RoundtableState,
    SeatExecutor,
)
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import Base, MafRoundtableRunRow, MafRoundtableSessionRow

MANAGER = "manager-1"
SEATS = ["seat-a", "seat-b"]


class _FakeCtx:
    def __init__(self) -> None:
        self.messages: list[object] = []
        self.outputs: list[object] = []

    async def send_message(self, msg: object) -> None:
        self.messages.append(msg)

    async def yield_output(self, msg: object) -> None:
        self.outputs.append(msg)


@pytest_asyncio.fixture
async def svc():
    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)
    store = make_maf_roundtable_store(sf)
    service = make_maf_service(store)
    now = datetime(2026, 1, 1)
    async with sf() as session:
        session.add(MafRoundtableSessionRow(
            id="sess-1", user_id="u1", title="t", task_text="分析市场",
            mode="structured", status="running", write_policy="auto",
            manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
        ))
        session.add(MafRoundtableRunRow(
            id="run-1", session_id="sess-1", status="running",
            created_at=now, updated_at=now,
        ))
        await session.commit()
    yield service
    await engine.dispose()


def _state() -> RoundtableState:
    return RoundtableState(
        session_id="sess-1", run_id="run-1", task_text="分析市场",
        mode="structured", model_name=None, reasoning_effort=None,
        max_rounds=8, write_policy="auto", manager_agent_id=MANAGER,
        seat_agent_ids=SEATS,
        thread_map={MANAGER: "t-mgr", "seat-a": "t-a", "seat-b": "t-b"},
        capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True,
    )


def _resp(content: str) -> SeatResponse:
    return SeatResponse(
        agent_id="x", turn_index=0, content=content,
        tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1},
    )


class _RecordingAdapter:
    """Fake adapter that (a) records every run_seat_turn's execution_stage and
    (b) writes a REAL seat-run row for each call so the four-tuple coexistence
    invariant is checked against the actual DB."""

    def __init__(self, service) -> None:
        self._service = service
        self.calls: list[tuple[str, str, int, str]] = []  # (agent, turn, stage)
        self._seat_outputs: dict[tuple[str, str], str] = {}  # (agent, stage) -> content
        self._user_id = "u1"
        self._emit_critical = AsyncMock()
        self._emit = AsyncMock()

    def set_output(self, agent_id: str, execution_stage: str, content: str) -> None:
        self._seat_outputs[(agent_id, execution_stage)] = content

    async def run_seat_turn(self, **kwargs):
        agent_id = kwargs["agent_id"]
        turn_index = kwargs["turn_index"]
        stage = kwargs.get("execution_stage", "expert")
        self.calls.append((agent_id, turn_index, stage))
        # Persist a seat-run row so coexistence is verified against the real DB.
        await self._service.store.seat_runs.create_seat_run(
            maf_run_id=kwargs["run_id"], agent_id=agent_id,
            turn_index=turn_index, execution_stage=stage,
        )
        content = self._seat_outputs.get((agent_id, stage), '{"summary":"ok"}')
        return _resp(content)


# ── 1. Full structured run: manager stages coexist ───────────────────────────


@pytest.mark.asyncio
async def test_full_structured_run_manager_stages_coexist(svc):
    """A complete planner→experts→cross_review→approval→finalizer run must
    persist manager planner/cross_review/finalizer rows that ALL share the same
    business round (turn_index) but differ by execution_stage — no collision."""
    adapter = _RecordingAdapter(svc)
    # Manager emits valid JSON at each stage.
    adapter.set_output(MANAGER, "planner", "分工：seat-a 负责需求，seat-b 负责竞品")
    adapter.set_output(MANAGER, "cross_review", json.dumps({
        "summary": "审查通过",
        "evidence_ok": True,
        "confirmed_candidates": [],
    }))
    adapter.set_output(MANAGER, "finalizer", "最终结论：市场看好")
    # Seats emit valid structured output.
    for s in SEATS:
        adapter.set_output(s, "expert", json.dumps({"summary": f"{s} 结论", "findings": ["f"]}))

    state = _state()
    ctx = _FakeCtx()
    await PlannerExecutor(adapter, state).handle("task", ctx)
    for s in SEATS:
        await SeatExecutor(adapter, state, s).handle({}, _FakeCtx())
    await CrossReviewExecutor(adapter, state).handle({}, _FakeCtx())
    await ApprovalGateExecutor(adapter, state, svc).handle({}, _FakeCtx())
    await FinalizerExecutor(adapter, state).handle({}, _FakeCtx())

    # Assert every manager stage produced a distinct seat-run row, all sharing
    # turn_index (planner=0, cross_review/finalizer advance current_round; the
    # invariant is that the SAME (agent) has multiple distinct stages).
    rows = await svc.store.seat_runs.get_by_maf_run("run-1")
    mgr_rows = [r for r in rows if r["agent_id"] == MANAGER]
    mgr_stages = {r["execution_stage"] for r in mgr_rows}
    assert {"planner", "cross_review", "finalizer"}.issubset(mgr_stages), \
        f"manager stages missing; got {mgr_stages}"
    # No duplicate four-tuple (the unique constraint would have raised).
    seen = set()
    for r in mgr_rows:
        key = (r["maf_run_id"], r["agent_id"], r["turn_index"], r["execution_stage"])
        assert key not in seen, f"duplicate four-tuple {key}"
        seen.add(key)
    # The final result was produced.
    assert state.final_result is not None
    assert state.status == "completed"


# ── 2. expert + format_repair coexist ────────────────────────────────────────


@pytest.mark.asyncio
async def test_expert_format_repair_coexist(svc):
    """A seat that emits invalid JSON triggers exactly one ``format_repair``
    DeerFlow call; both the expert and the format_repair rows persist for the
    same (agent, turn) without colliding."""
    adapter = _RecordingAdapter(svc)
    # First expert call: invalid JSON → triggers repair. The format_repair call
    # (stage='format_repair') returns valid JSON.
    adapter.set_output("seat-a", "expert", "not-json-at-all")
    adapter.set_output("seat-a", "format_repair", json.dumps({"summary": "repaired"}))

    state = _state()
    state.current_round = 1
    state.plan = {"assignments": {"seat-a": {"instruction": "x", "role": "A"}}}

    await SeatExecutor(adapter, state, "seat-a").handle({}, _FakeCtx())

    stage_calls = [(a, t, s) for (a, t, s) in adapter.calls if a == "seat-a"]
    stages = [s for _, _, s in stage_calls]
    assert "expert" in stages and "format_repair" in stages
    # Exactly one repair (not a retry loop).
    assert stages.count("format_repair") == 1

    rows = await svc.store.seat_runs.get_by_maf_run("run-1")
    seat_a_stages = {r["execution_stage"] for r in rows if r["agent_id"] == "seat-a"}
    assert seat_a_stages == {"expert", "format_repair"}


# ── 3. cross_review + cross_review_repair coexist ────────────────────────────


@pytest.mark.asyncio
async def test_cross_review_repair_coexist(svc):
    """An invalid cross-review output triggers exactly one ``cross_review_repair``
    DeerFlow call; cross_review and cross_review_repair rows coexist for the
    manager."""
    adapter = _RecordingAdapter(svc)
    adapter.set_output(MANAGER, "cross_review", "totally not json")
    adapter.set_output(MANAGER, "cross_review_repair", json.dumps({
        "summary": "repaired review", "confirmed_candidates": [],
    }))

    state = _state()
    # Pretend experts already ran so the cross-review barrier sees outcomes.
    state.seat_outcomes = {s: {"status": "completed"} for s in SEATS}
    state.current_round = 1

    await CrossReviewExecutor(adapter, state).handle({}, _FakeCtx())

    mgr_stages = [s for a, _, s in adapter.calls if a == MANAGER]
    assert "cross_review" in mgr_stages and "cross_review_repair" in mgr_stages
    assert mgr_stages.count("cross_review_repair") == 1

    rows = await svc.store.seat_runs.get_by_maf_run("run-1")
    mgr_row_stages = {r["execution_stage"] for r in rows if r["agent_id"] == MANAGER}
    assert {"cross_review", "cross_review_repair"}.issubset(mgr_row_stages)

    # The repaired review was applied (deterministic degradation did NOT happen).
    assert state.cross_review_result is not None
    assert state.cross_review_result.get("schema_status") == "repaired"
    assert state.cross_review_result.get("summary") == "repaired review"


# ── 4. manager present in participants-style config ──────────────────────────


@pytest.mark.asyncio
async def test_manager_role_wired_through_state(svc):
    """Sanity: RoundtableState carries manager_agent_id and the planner uses it
    for the planner seat-turn (not a seat id)."""
    state = _state()
    assert state.manager_agent_id == MANAGER
    assert MANAGER not in state.seat_agent_ids

    adapter = _RecordingAdapter(svc)
    adapter.set_output(MANAGER, "planner", "plan")
    ctx = _FakeCtx()
    await PlannerExecutor(adapter, state).handle("task", ctx)

    planner_calls = [(a, s) for a, _, s in adapter.calls if s == "planner"]
    assert planner_calls == [(MANAGER, "planner")], \
        "planner stage must run as the MANAGER, not a seat"


# ── P1-3 / P1-4: Runner-level full chain tests ──────────────────────────────


@pytest_asyncio.fixture
async def v2_svc(tmp_path):
    """Service with a V2 session: manager + 3 candidate seats, dispatch_contract_enabled."""
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'v2runner.db'}", echo=False)
    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)
    service = make_maf_service(store)
    now = datetime(2026, 1, 1)
    async with sf() as session:
        session.add(MafRoundtableSessionRow(
            id="v2-sess", user_id="u1", title="v2", task_text="分析市场趋势",
            mode="structured", status="pending", write_policy="auto",
            manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
            task_payload=json.dumps({
                "maf_v2": {
                    "dispatch_contract_enabled": True,
                    "candidate_set": {
                        "manager_agent_id": MANAGER,
                        "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"],
                        "mode": "structured",
                        "max_rounds": 8,
                    },
                },
            }),
        ))
        session.add(MafRoundtableRunRow(
            id="v2-run", session_id="v2-sess", status="pending",
            created_at=now, updated_at=now,
        ))
        # Persist ALL candidates (simulating old V2 behavior).
        from deerflow.persistence.maf_roundtable.model import MafRoundtableParticipantRow
        for i, (agent_id, role) in enumerate([(MANAGER, "manager"), ("seat-a", "seat"), ("seat-b", "seat"), ("seat-c", "seat")]):
            session.add(MafRoundtableParticipantRow(
                id=f"p-{i}", session_id="v2-sess", agent_id=agent_id,
                role=role, sort_order=i, created_at=now, updated_at=now,
            ))
        await session.commit()
    yield service
    await engine.dispose()


def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict):
    """Create a WorkflowRunner with a mocked adapter for V2 tests."""
    from types import SimpleNamespace
    from unittest.mock import AsyncMock

    from app.maf_roundtable.workflow_runner import WorkflowRunner

    app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=v2_svc, checkpointer=None))
    runner = WorkflowRunner(app, service=v2_svc, user_id="u1")

    calls: list[dict] = []

    async def _run_seat(**kwargs):
        calls.append(kwargs)
        stage = kwargs.get("execution_stage", "")
        agent_id = kwargs["agent_id"]
        if stage == "manager_dispatch":
            return SeatResponse(agent_id=agent_id, turn_index=0, content=json.dumps(dispatch_decision), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "direct_answer":
            return SeatResponse(agent_id=agent_id, turn_index=0, content="manager direct answer", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "expert":
            return SeatResponse(agent_id=agent_id, turn_index=1, content=json.dumps({"summary": f"{agent_id} result", "findings": ["f"]}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "cross_review":
            return SeatResponse(agent_id=agent_id, turn_index=2, content=json.dumps({"summary": "ok", "confirmed_candidates": []}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "finalizer":
            return SeatResponse(agent_id=agent_id, turn_index=3, content="final report", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        return SeatResponse(agent_id=agent_id, turn_index=0, content='{"summary":"ok"}', tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})

    ensure_threads_calls: list[list[str]] = []

    async def _ensure_threads(session_id, run_id, participants):
        agent_ids = [p["agent_id"] for p in participants]
        ensure_threads_calls.append(agent_ids)
        return {p["agent_id"]: f"thread-{p['agent_id']}" for p in participants}

    monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat))
    monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(side_effect=_ensure_threads))
    monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock())
    monkeypatch.setattr(runner._adapter, "_emit", AsyncMock())
    return runner, calls, ensure_threads_calls


@pytest.mark.asyncio
async def test_runner_parallel_consultation_only_selected_seats(v2_svc, monkeypatch):
    """parallel_consultation: initial ensure_threads only gets manager;
    after dispatch, only selected seats A/B get threads; C never appears."""
    pytest.importorskip("agent_framework")

    dispatch_decision = {
        "intent_type": "parallel_consultation",
        "selected_seat_agent_ids": ["seat-a", "seat-b"],
        "assignments": [
            {"agent_id": "seat-a", "task_goal": "market analysis", "delivery_requirements": ["data"], "order": 1},
            {"agent_id": "seat-b", "task_goal": "competitor review", "delivery_requirements": ["sources"], "order": 2},
        ],
        "execution_mode": "parallel",
        "convergence_path": ["manager synthesis"],
        "reason": "two experts needed",
    }

    runner, calls, ensure_threads_calls = _make_v2_runner(v2_svc, monkeypatch, dispatch_decision=dispatch_decision)

    session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
    run = await v2_svc.store.runs.get_run("v2-run")
    await runner.execute(session, run)

    # 1. First ensure_threads call: only manager.
    assert len(ensure_threads_calls) >= 1
    assert ensure_threads_calls[0] == [MANAGER], \
        f"First ensure_threads must only get manager, got {ensure_threads_calls[0]}"

    # 2. seat-c never appears in any ensure_threads call.
    all_ensured = [aid for call in ensure_threads_calls for aid in call]
    assert "seat-c" not in all_ensured, "seat-c must never receive a thread"

    # 3. seat-a and seat-b were executed (expert stage).
    expert_agents = [c["agent_id"] for c in calls if c.get("execution_stage") == "expert"]
    assert "seat-a" in expert_agents
    assert "seat-b" in expert_agents
    assert "seat-c" not in expert_agents

    # 4. Run completed.
    final_run = await v2_svc.store.runs.get_run("v2-run")
    assert final_run["status"] == "completed"


@pytest.mark.asyncio
async def test_runner_direct_answer_zero_seats(v2_svc, monkeypatch):
    """direct_answer: manager_only — zero seat participants, zero seat threads,
    zero seat runs; final answer comes from manager."""
    pytest.importorskip("agent_framework")

    dispatch_decision = {
        "intent_type": "direct_answer",
        "selected_seat_agent_ids": [],
        "assignments": [],
        "execution_mode": "manager_only",
        "convergence_path": ["manager answer"],
        "reason": "simple question",
    }

    runner, calls, ensure_threads_calls = _make_v2_runner(v2_svc, monkeypatch, dispatch_decision=dispatch_decision)

    session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
    run = await v2_svc.store.runs.get_run("v2-run")
    await runner.execute(session, run)

    # 1. Only manager gets a thread.
    all_ensured = [aid for call in ensure_threads_calls for aid in call]
    assert all_ensured == [MANAGER], f"Only manager should get a thread, got {all_ensured}"

    # 2. No seat was executed (no expert stage calls).
    expert_calls = [c for c in calls if c.get("execution_stage") == "expert"]
    assert expert_calls == [], "No seat should execute in direct_answer mode"

    # 3. Manager produced the answer via direct_answer stage.
    direct_calls = [c for c in calls if c.get("execution_stage") == "direct_answer"]
    assert len(direct_calls) == 1
    assert direct_calls[0]["agent_id"] == MANAGER

    # 4. Run completed.
    final_run = await v2_svc.store.runs.get_run("v2-run")
    assert final_run["status"] == "completed"
    # 5. Zero seat runs in DB.
    seat_runs = await v2_svc.store.seat_runs.get_by_maf_run("v2-run")
    expert_rows = [r for r in seat_runs if r["execution_stage"] == "expert"]
    assert expert_rows == []


@pytest.mark.asyncio
async def test_runner_clarification_waiting_input(v2_svc, monkeypatch):
    """clarification: manager returns clarification — no seats started,
    run/session end as waiting_input."""
    pytest.importorskip("agent_framework")

    dispatch_decision = {
        "intent_type": "clarification",
        "selected_seat_agent_ids": [],
        "assignments": [],
        "execution_mode": "manager_only",
        "convergence_path": ["clarification"],
        "reason": "task unclear",
        "clarification_prompt": "请明确分析的目标市场和时间范围。",
    }

    runner, calls, ensure_threads_calls = _make_v2_runner(v2_svc, monkeypatch, dispatch_decision=dispatch_decision)

    session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
    run = await v2_svc.store.runs.get_run("v2-run")
    await runner.execute(session, run)

    # 1. Only manager gets a thread.
    all_ensured = [aid for call in ensure_threads_calls for aid in call]
    assert all_ensured == [MANAGER]

    # 2. No seat executed.
    expert_calls = [c for c in calls if c.get("execution_stage") == "expert"]
    assert expert_calls == []

    # 3. Run and session are waiting_input.
    final_run = await v2_svc.store.runs.get_run("v2-run")
    assert final_run["status"] == "waiting_input"
    final_session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
    assert final_session["status"] == "waiting_input"

    # 4. Checkpoint preserves the clarification prompt.
    checkpoint = final_run.get("maf_checkpoint") or {}
    assert checkpoint.get("intent_type") == "clarification"
    plan = checkpoint.get("dispatch_plan") or {}
    assert plan.get("clarification_prompt") == "请明确分析的目标市场和时间范围。"


# ── P1-3: _participants_requiring_threads unit tests ────────────────────────


def test_participants_requiring_threads_legacy_returns_all():
    """Legacy MAF sessions without V2 contract keep original behavior."""
    from types import SimpleNamespace

    from app.maf_roundtable.workflow_runner import WorkflowRunner

    app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=None, checkpointer=None))
    runner = WorkflowRunner.__new__(WorkflowRunner)
    runner._app = app
    runner._service = None
    runner._user_id = "u1"

    participants = [
        {"agent_id": "manager", "role": "manager"},
        {"agent_id": "seat-a", "role": "seat"},
        {"agent_id": "seat-b", "role": "seat"},
    ]
    result = runner._participants_requiring_threads(
        participants,
        dispatch_contract_enabled=False,
        dispatch_plan=None,
        intent_type=None,
    )
    assert result == participants


def test_participants_requiring_threads_v2_no_plan_only_manager():
    """V2 session without a validated plan: only manager gets threads."""
    from types import SimpleNamespace

    from app.maf_roundtable.workflow_runner import WorkflowRunner

    runner = WorkflowRunner.__new__(WorkflowRunner)
    runner._app = None
    runner._service = None
    runner._user_id = "u1"

    participants = [
        {"agent_id": "manager", "role": "manager"},
        {"agent_id": "seat-a", "role": "seat"},
        {"agent_id": "seat-b", "role": "seat"},
        {"agent_id": "seat-c", "role": "seat"},
    ]
    result = runner._participants_requiring_threads(
        participants,
        dispatch_contract_enabled=True,
        dispatch_plan=None,
        intent_type=None,
    )
    assert [p["agent_id"] for p in result] == ["manager"]


def test_participants_requiring_threads_v2_direct_answer_only_manager():
    """V2 with validated direct_answer plan: only manager."""
    from app.maf_roundtable.workflow_runner import WorkflowRunner

    runner = WorkflowRunner.__new__(WorkflowRunner)
    runner._app = None
    runner._service = None
    runner._user_id = "u1"

    participants = [
        {"agent_id": "manager", "role": "manager"},
        {"agent_id": "seat-a", "role": "seat"},
    ]
    result = runner._participants_requiring_threads(
        participants,
        dispatch_contract_enabled=True,
        dispatch_plan={"intent_type": "direct_answer", "execution_mode": "manager_only"},
        intent_type="direct_answer",
    )
    assert [p["agent_id"] for p in result] == ["manager"]


def test_participants_requiring_threads_v2_selected_seats():
    """V2 with validated parallel plan: manager + selected seats only."""
    from app.maf_roundtable.workflow_runner import WorkflowRunner

    runner = WorkflowRunner.__new__(WorkflowRunner)
    runner._app = None
    runner._service = None
    runner._user_id = "u1"

    participants = [
        {"agent_id": "manager", "role": "manager"},
        {"agent_id": "seat-a", "role": "seat"},
        {"agent_id": "seat-b", "role": "seat"},
        {"agent_id": "seat-c", "role": "seat"},
    ]
    result = runner._participants_requiring_threads(
        participants,
        dispatch_contract_enabled=True,
        dispatch_plan={
            "intent_type": "parallel_consultation",
            "execution_mode": "parallel",
            "seat_agent_ids": ["seat-a", "seat-b"],
        },
        intent_type="parallel_consultation",
    )
    agent_ids = [p["agent_id"] for p in result]
    assert "manager" in agent_ids
    assert "seat-a" in agent_ids
    assert "seat-b" in agent_ids
    assert "seat-c" not in agent_ids