"""Phase-3 acceptance tests for the fact ledger + cross-review wiring.

Covers P1-2 (candidate_facts separation), P1-3 (approval ledger integrity),
and P1-1 (CrossReviewOutput validation + deterministic degradation).

These are pure-unit tests over the fact_ledger / seat_output / workflow_executors
helpers — no DB, no real DeerFlow.
"""

from __future__ import annotations

import json
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from app.maf_roundtable.fact_ledger import (
    candidate_id_map,
    confirm_candidate_facts,
    empty_ledger,
    merge_seat_output,
    record_action_transition,
)
from app.maf_roundtable.seat_output import parse_cross_review_output

# ── P1-2: seat findings → candidate_facts, NOT confirmed_facts ─────────────


def test_seat_findings_land_in_candidate_facts_not_confirmed():
    """A seat's `findings` must NEVER be auto-promoted to confirmed_facts —
    weak model output is a claim, not a fact."""
    ledger = empty_ledger()
    merge_seat_output(ledger, {"findings": ["市场增长10%", "竞品A领先"]}, agent_id="seat-1")

    assert len(ledger["candidate_facts"]) == 2
    assert ledger["confirmed_facts"] == [], "seat findings must not be confirmed facts"

    # Each candidate gets a server-generated candidate_id.
    for c in ledger["candidate_facts"]:
        assert "candidate_id" in c and c["candidate_id"]
        assert c["confirmed"] is False
        assert c["agent_id"] == "seat-1"
        assert c["source"] == "seat_finding"


def test_candidate_facts_promoted_only_after_cross_review_confirmation():
    """Promotion to confirmed_facts happens only via confirm_candidate_facts,
    and only for ids that actually exist in candidate_facts AND cite evidence
    the ledger has backing for."""
    from app.maf_roundtable.fact_ledger import record_tool_result

    ledger = empty_ledger()
    merge_seat_output(ledger, {"findings": ["事实A", "事实B"]}, agent_id="seat-1")
    ids = [c["candidate_id"] for c in ledger["candidate_facts"]]
    assert len(ids) == 2

    # Record real evidence so the confirmation can cite it (Phase-3 P1-1).
    record_tool_result(
        ledger, tool_name="web_search", status="completed",
        agent_id="seat-1", turn_index=1,
    )

    # Model tries to confirm one real id + one HALLUCINATED id, citing the
    # real evidence for both.
    tool_ref = "tool:web_search:1"
    confirm_candidate_facts(
        ledger,
        confirmed_candidate_ids=[ids[0], "nonexistent-id"],
        confirmed_by="manager-1",
        evidence_refs=[tool_ref],
    )

    assert len(ledger["confirmed_facts"]) == 1  # only the real one
    assert ledger["confirmed_facts"][0]["candidate_id"] == ids[0]
    assert ledger["confirmed_facts"][0]["confirmed_by"] == "manager-1"
    assert ledger["confirmed_facts"][0]["source"] == "cross_review_confirmed"
    # The confirmed candidate is marked confirmed=True in place.
    assert candidate_id_map(ledger)[ids[0]]["confirmed"] is True
    assert candidate_id_map(ledger)[ids[1]]["confirmed"] is False


def test_candidate_not_promoted_without_ledger_backed_evidence():
    """Phase-3 P1-1: confirmation requires evidence that resolves to a ledger
    entry. A candidate cited with an invented evidence ref (not recorded in the
    ledger) is REJECTED and stays a candidate."""
    from app.maf_roundtable.fact_ledger import confirm_candidate_facts

    ledger = empty_ledger()
    merge_seat_output(ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    cid = ledger["candidate_facts"][0]["candidate_id"]

    # No evidence recorded in the ledger; the model cites an invented ref.
    confirm_candidate_facts(
        ledger,
        confirmed_candidate_ids=[cid],
        confirmed_by="manager-1",
        evidence_refs=["invented-evidence-not-in-ledger"],
    )

    assert ledger["confirmed_facts"] == [], "candidate promoted with non-ledger evidence"
    assert candidate_id_map(ledger)[cid]["confirmed"] is False


def test_candidate_not_promoted_with_no_evidence_at_all():
    """Phase-3 P1-1: a confirmation that supplies no evidence_refs is rejected."""
    from app.maf_roundtable.fact_ledger import confirm_candidate_facts

    ledger = empty_ledger()
    merge_seat_output(ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    cid = ledger["candidate_facts"][0]["candidate_id"]

    confirm_candidate_facts(
        ledger,
        confirmed_candidate_ids=[cid],
        confirmed_by="manager-1",
        evidence_refs=None,
    )

    assert ledger["confirmed_facts"] == []
    assert candidate_id_map(ledger)[cid]["confirmed"] is False


def test_confirm_is_idempotent():
    """Confirming the same candidate_id twice does not duplicate it."""
    from app.maf_roundtable.fact_ledger import record_tool_result

    ledger = empty_ledger()
    merge_seat_output(ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    cid = ledger["candidate_facts"][0]["candidate_id"]
    record_tool_result(
        ledger, tool_name="search", status="completed",
        agent_id="seat-1", turn_index=1,
    )
    ref = "tool:search:1"
    confirm_candidate_facts(ledger, confirmed_candidate_ids=[cid], confirmed_by="m1", evidence_refs=[ref])
    confirm_candidate_facts(ledger, confirmed_candidate_ids=[cid], confirmed_by="m1", evidence_refs=[ref])
    assert len(ledger["confirmed_facts"]) == 1


def test_empty_ledger_has_candidate_partition():
    ledger = empty_ledger()
    assert "candidate_facts" in ledger and ledger["candidate_facts"] == []
    assert "confirmed_facts" in ledger


# ── P1-3: approval ledger integrity ────────────────────────────────────────


def test_pending_action_transitions_on_reject():
    """A pending action that is rejected moves to the failed/unknown partition
    with status 'rejected' — it must not stay in pending and must not appear in
    executed."""
    ledger = empty_ledger()
    record_action_transition(ledger, action_id="act-1", tool_name="write_file",
                             status="pending", agent_id="seat-1")
    assert len(ledger["pending_actions"]) == 1

    record_action_transition(ledger, action_id="act-1", tool_name="write_file",
                             status="rejected", agent_id="seat-1")

    # Removed from pending.
    assert not any(p.get("action_id") == "act-1" for p in ledger["pending_actions"])
    # Present in failed/unknown with status rejected.
    failed = [f for f in ledger["failed_or_unknown_actions"] if f.get("action_id") == "act-1"]
    assert len(failed) == 1
    assert failed[0]["status"] == "rejected"
    # NOT in executed.
    assert not any(e.get("action_id") == "act-1" for e in ledger["executed_actions"])


def test_pending_action_transitions_on_execute():
    """A pending action that executes successfully moves pending → executed."""
    ledger = empty_ledger()
    record_action_transition(ledger, action_id="act-2", tool_name="create_ticket",
                             status="pending", agent_id="seat-1")
    record_action_transition(ledger, action_id="act-2", tool_name="create_ticket",
                             status="executed", agent_id="seat-1")
    assert not any(p.get("action_id") == "act-2" for p in ledger["pending_actions"])
    assert any(e.get("action_id") == "act-2" and e.get("status") == "executed"
               for e in ledger["executed_actions"])
    assert not any(f.get("action_id") == "act-2" for f in ledger["failed_or_unknown_actions"])


def test_failed_action_never_in_executed():
    """A failed/unknown action must never leak into executed_actions."""
    ledger = empty_ledger()
    record_action_transition(ledger, action_id="act-3", tool_name="write_file",
                             status="pending", agent_id="seat-1")
    for terminal in ("failed", "unknown"):
        record_action_transition(ledger, action_id="act-3", tool_name="write_file",
                                 status=terminal, agent_id="seat-1")
        assert not any(e.get("action_id") == "act-3" for e in ledger["executed_actions"]), \
            f"{terminal} action leaked into executed_actions"


# ── P1-1: CrossReviewOutput parsing ────────────────────────────────────────


def test_parse_cross_review_output_valid():
    payload = {
        "summary": "审查通过",
        "evidence_ok": True,
        "gaps": ["缺少 X 证据"],
        "conflicts": [],
        "tool_failures": [],
        "write_status_notes": ["待审批1个"],
        "recommendations": ["补充X"],
        "confirmed_candidate_ids": ["c1", "c2"],
    }
    model, errors = parse_cross_review_output(json.dumps(payload))
    assert model is not None
    assert errors == []
    assert model.summary == "审查通过"
    assert model.confirmed_candidate_ids == ["c1", "c2"]


def test_parse_cross_review_confirmed_candidates_requires_evidence():
    """Phase-3 P1-1: each ConfirmedCandidate must carry ≥1 evidence ref — an
    empty list is a schema violation (so the executor repairs instead of
    silently accepting a bare-id confirmation)."""
    payload = {
        "summary": "ok",
        "confirmed_candidates": [{"candidate_id": "c1", "evidence_refs": []}],
    }
    model, errors = parse_cross_review_output(json.dumps(payload))
    assert model is None
    assert any("evidence_refs" in e or "confirmed_candidates" in e for e in errors)


def test_parse_cross_review_evidence_ok_defaults_false():
    """Phase-3 P1-1: omitting evidence_ok must default to False — a model that
    forgets the field must never be treated as 'evidence sufficient'."""
    model, errors = parse_cross_review_output(json.dumps({"summary": "ok"}))
    assert model is not None
    assert errors == []
    assert model.evidence_ok is False


def test_parse_cross_review_output_invalid_json_returns_errors():
    model, errors = parse_cross_review_output("not json at all")
    assert model is None
    assert errors  # non-empty readable errors


def test_parse_cross_review_output_missing_summary_invalid():
    """summary is required (min_length=1) — a missing summary must fail
    validation so the CrossReviewExecutor repairs rather than accepting it."""
    model, errors = parse_cross_review_output(json.dumps({"gaps": ["x"]}))
    assert model is None
    assert any("summary" in e for e in errors)


# ── P1-1: CrossReviewExecutor deterministic degradation ────────────────────
#
# These exercise the executor's _apply_cross_review_output path with a fake
# adapter, proving: (a) valid output confirms candidates; (b) invalid output
# that the repair also cannot fix degrades deterministically WITHOUT pasting
# raw JSON into summary.


def _make_state():
    """Minimal RoundtableState for CrossReviewExecutor tests."""
    from app.maf_roundtable.workflow_executors import RoundtableState

    return RoundtableState(
        session_id="s1", run_id="r1", task_text="t", mode="structured",
        model_name=None, reasoning_effort=None, max_rounds=8,
        write_policy="confirm_high_risk", manager_agent_id="mgr",
        seat_agent_ids=["seat-1"], thread_map={"mgr": "t-mgr"},
    )


@pytest.mark.asyncio
async def test_cross_review_valid_output_confirms_candidates():
    from app.maf_roundtable.fact_ledger import record_tool_result
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    merge_seat_output(state.fact_ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    candidate_id = state.fact_ledger["candidate_facts"][0]["candidate_id"]
    # Record evidence so the confirmation can bind it (Phase-3 P1-1).
    record_tool_result(
        state.fact_ledger, tool_name="search", status="completed",
        agent_id="seat-1", turn_index=1,
    )
    tool_ref = "tool:search:1"

    executor = CrossReviewExecutor(adapter=SimpleNamespace(), state=state)
    review: dict = {}
    # Structured confirmed_candidates path: each candidate cites ledger evidence.
    raw = json.dumps({
        "summary": "ok",
        "evidence_ok": True,
        "confirmed_candidates": [{"candidate_id": candidate_id, "evidence_refs": [tool_ref]}],
    })
    await executor._apply_cross_review_output(review, raw, "mgr")

    assert review["source"] == "manager"
    assert review["schema_status"] == "ok"
    assert review["summary"] == "ok"
    # The candidate was promoted to confirmed_facts.
    assert len(state.fact_ledger["confirmed_facts"]) == 1
    assert state.fact_ledger["confirmed_facts"][0]["candidate_id"] == candidate_id
    # The public confirmation field contains only the server-accepted item;
    # the model request is retained separately for audit.
    assert review["confirmed_candidate_ids"] == [candidate_id]
    assert review["confirmed_candidates"] == [
        {"candidate_id": candidate_id, "evidence_refs": [tool_ref]}
    ]
    assert review["requested_confirmations"] == review["confirmed_candidates"]


@pytest.mark.asyncio
async def test_cross_review_rejects_candidate_with_invented_evidence():
    """Phase-3 P1-1: a confirmation that cites evidence NOT in the ledger is
    rejected server-side. The candidate stays unconfirmed, and the rejection is
    surfaced in the review so the summary can explain it."""
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    merge_seat_output(state.fact_ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    candidate_id = state.fact_ledger["candidate_facts"][0]["candidate_id"]

    executor = CrossReviewExecutor(adapter=SimpleNamespace(), state=state)
    review: dict = {}
    raw = json.dumps({
        "summary": "ok",
        "evidence_ok": True,
        "confirmed_candidates": [
            {"candidate_id": candidate_id, "evidence_refs": ["invented-not-in-ledger"]},
        ],
    })
    await executor._apply_cross_review_output(review, raw, "mgr")

    # Not promoted.
    assert state.fact_ledger["confirmed_facts"] == []
    # Rejection recorded.
    assert review.get("confirmation_rejections"), "rejection not surfaced in review"
    rej = review["confirmation_rejections"][0]
    assert rej["candidate_id"] == candidate_id
    assert rej["reason"] == "evidence_not_in_ledger"
    assert review["confirmed_candidates"] == []
    assert review["confirmed_candidate_ids"] == []
    assert review["requested_confirmations"] == [
        {"candidate_id": candidate_id, "evidence_refs": ["invented-not-in-ledger"]}
    ]


@pytest.mark.asyncio
async def test_cross_review_invalid_then_repair_succeeds():
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    # Fake adapter: repair turn returns valid JSON.
    good_repair = json.dumps({"summary": "repaired ok", "confirmed_candidate_ids": []})
    fake_adapter = SimpleNamespace(
        run_seat_turn=AsyncMock(return_value=SimpleNamespace(content=good_repair)),
    )
    executor = CrossReviewExecutor(adapter=fake_adapter, state=state)
    review: dict = {}
    # First output is garbage; repair returns valid.
    await executor._apply_cross_review_output(review, "totally not json", "mgr")

    assert review["source"] == "manager"
    assert review["schema_status"] == "repaired"
    assert review["summary"] == "repaired ok"
    fake_adapter.run_seat_turn.assert_awaited_once()


@pytest.mark.asyncio
async def test_cross_review_invalid_and_repair_fails_degrades_deterministically():
    """When both the original and the repair output are unparseable, the
    review MUST degrade deterministically: raw JSON is never pasted into
    summary, and the review is marked schema_invalid."""
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    # Repair also returns garbage.
    fake_adapter = SimpleNamespace(
        run_seat_turn=AsyncMock(return_value=SimpleNamespace(content="still garbage")),
    )
    executor = CrossReviewExecutor(adapter=fake_adapter, state=state)
    review: dict = {}
    await executor._apply_cross_review_output(review, "garbage first output", "mgr")

    assert review["source"] == "deterministic"
    assert review["schema_status"] == "schema_invalid"
    assert review.get("raw_response_ignored") is True
    # summary must NOT be the raw model JSON.
    assert review.get("summary", "") != "garbage first output"


# ── P1-2: evidence_ok=False blocks all promotion ────────────────────────────


@pytest.mark.asyncio
async def test_evidence_ok_false_blocks_promotion_even_with_valid_refs():
    """Phase-3 P1-2: when the model says evidence_ok=False, no candidate may
    be promoted even if it cites valid ledger-backed evidence refs. The
    attempted confirmations are recorded as rejections."""
    from app.maf_roundtable.fact_ledger import record_tool_result
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    merge_seat_output(state.fact_ledger, {"findings": ["事实A"]}, agent_id="seat-1")
    candidate_id = state.fact_ledger["candidate_facts"][0]["candidate_id"]
    record_tool_result(
        state.fact_ledger, tool_name="search", status="completed",
        agent_id="seat-1", turn_index=1,
    )
    tool_ref = "tool:search:1"

    executor = CrossReviewExecutor(adapter=SimpleNamespace(), state=state)
    review: dict = {}
    raw = json.dumps({
        "summary": "证据不充分",
        "evidence_ok": False,
        "confirmed_candidates": [
            {"candidate_id": candidate_id, "evidence_refs": [tool_ref]},
        ],
    })
    await executor._apply_cross_review_output(review, raw, "mgr")

    assert state.fact_ledger["confirmed_facts"] == []
    assert candidate_id_map(state.fact_ledger)[candidate_id]["confirmed"] is False
    assert review.get("confirmation_rejections")
    rej = review["confirmation_rejections"][0]
    assert rej["candidate_id"] == candidate_id
    assert rej["reason"] == "review_did_not_confirm_evidence"
    assert review["confirmed_candidates"] == []
    assert review["confirmed_candidate_ids"] == []
    assert review["requested_confirmations"] == [
        {"candidate_id": candidate_id, "evidence_refs": [tool_ref]}
    ]


@pytest.mark.asyncio
async def test_evidence_ok_true_allows_promotion_with_valid_refs():
    """Phase-3 P1-2: when evidence_ok=True AND evidence refs are valid, the
    candidate IS promoted."""
    from app.maf_roundtable.fact_ledger import record_tool_result
    from app.maf_roundtable.workflow_executors import CrossReviewExecutor

    state = _make_state()
    merge_seat_output(state.fact_ledger, {"findings": ["事实B"]}, agent_id="seat-1")
    candidate_id = state.fact_ledger["candidate_facts"][0]["candidate_id"]
    record_tool_result(
        state.fact_ledger, tool_name="search", status="completed",
        agent_id="seat-1", turn_index=1,
    )
    tool_ref = "tool:search:1"

    executor = CrossReviewExecutor(adapter=SimpleNamespace(), state=state)
    review: dict = {}
    raw = json.dumps({
        "summary": "证据充分",
        "evidence_ok": True,
        "confirmed_candidates": [
            {"candidate_id": candidate_id, "evidence_refs": [tool_ref]},
        ],
    })
    await executor._apply_cross_review_output(review, raw, "mgr")

    assert len(state.fact_ledger["confirmed_facts"]) == 1
    assert state.fact_ledger["confirmed_facts"][0]["candidate_id"] == candidate_id