"""Stage 4 (R3-Q) runner-level tests: seat task quality loop + phase gate.

Drives the REAL workflow (QualitySeatTasksExecutor → QualityGateExecutor →
CrossReviewExecutor → ApprovalGateExecutor → FinalizerExecutor) through
WorkflowRunner.execute() with a fake adapter, proving the acceptance
criteria end-to-end against the real database:

- a good delivery is accepted and the run completes with quality_decision=accept;
- a short / evidence-free delivery is sent back, reworked up to the frozen cap,
  reassigned to a backup seat once, and finally ends as a traceable
  controlled_gap that is carried into the final report (never as a fact);
- an ``ask_user`` fallback pauses the run in waiting_input;
- a required task left non-terminal blocks the phase quality gate;
- a missing/invalid manager review NEVER auto-accepts: the server forces
  system reworks and exhausts into a controlled gap (source=system events).
"""

from __future__ import annotations

import json
from datetime import datetime
from types import SimpleNamespace
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 QualityGateExecutor, RoundtableState
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import (
    Base,
    MafRoundtableParticipantRow,
    MafRoundtableRunRow,
    MafRoundtableSessionRow,
)

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

# ``@TOOL_REF@`` is replaced per-turn with ``tool:kb_search:<turn>`` — the
# trusted evidence ref created by the seat's (fake) successful tool call, so a
# good delivery cites ONLY evidence the fact ledger actually backs (P1).
GOOD_DELIVERY = json.dumps({
    "summary": "结论：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，同时预留华南试点预算",
    "findings": ["依据一：华东区销售数据连续三个月增长", "依据二：竞品在华东投放低于均值"],
    "evidence_refs": ["@TOOL_REF@"],
    "open_questions": [],
    "self_check": "已覆盖全部验收标准",
})
SHORT_DELIVERY = json.dumps({"summary": "好", "findings": []})

DISPATCH_PARALLEL = {
    "intent_type": "parallel_consultation",
    "selected_seat_agent_ids": ["seat-a", "seat-b"],
    "assignments": [
        {"agent_id": "seat-a", "task_goal": "市场增长机会分析", "delivery_requirements": ["证据", "优先级"], "order": 1},
        {"agent_id": "seat-b", "task_goal": "竞品定价分析", "delivery_requirements": ["证据", "建议"], "order": 2},
    ],
    "execution_mode": "parallel",
    "convergence_path": ["manager synthesis"],
    "reason": "需要两位专家",
}


@pytest_asyncio.fixture
async def q_svc(tmp_path):
    """Service with a V2 session: manager + 3 candidates, quality gate ON."""
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'q4.db'}", echo=False)
    try:
        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="q-sess", user_id="u1", title="q", 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={
                    "maf_v2": {
                        "dispatch_contract_enabled": True,
                        "quality_gate_enabled": True,
                        "candidate_set": {
                            "manager_agent_id": MANAGER,
                            "candidate_seat_agent_ids": list(SEATS),
                            "mode": "structured",
                            "max_rounds": 8,
                        },
                    },
                },
            ))
            session.add(MafRoundtableRunRow(
                id="q-run", session_id="q-sess", status="pending",
                created_at=now, updated_at=now,
            ))
            for i, agent_id in enumerate([MANAGER, *SEATS]):
                session.add(MafRoundtableParticipantRow(
                    id=f"qp-{i}", session_id="q-sess", agent_id=agent_id,
                    role="manager" if agent_id == MANAGER else "seat",
                    sort_order=i, created_at=now, updated_at=now,
                ))
            await session.commit()
        yield service
    finally:
        await engine.dispose()


def _make_q_runner(
    q_svc,
    monkeypatch,
    *,
    deliveries: dict[str, list[str]],
    review_plan: list[dict[str, dict[str, object]]],
    fallback_policy: str = "reassign_then_gap",
    cross_review_ok: bool = True,
    emit_real: bool = False,
    config_overrides: dict | None = None,
):
    """Runner with a fake adapter driven by explicit delivery/review plans.

    ``deliveries``: agent_id → list of JSON payloads, one per delivery attempt.
    ``review_plan``: one entry per quality round → {agent_id: {"decision", ...}}.
    """
    from app.maf_roundtable.workflow_runner import WorkflowRunner

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

    calls: list[tuple[str, str, int]] = []
    stage_messages: dict[str, list[str]] = {}
    delivery_idx: dict[str, int] = {}
    review_round: list[int] = [0]

    cfg_values = {
        "seat_timeout_seconds": 5,
        "schema_repair_enabled": True,
        "quality_gate_enabled": True,
        "max_rework_attempts": 2,
        "max_reassignments": 1,
        "min_delivery_chars": 40,
        "min_delivery_evidence": 1,
        "default_fallback_policy": fallback_policy,
    }
    cfg_values.update(config_overrides or {})
    fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(**cfg_values))
    monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)

    async def _run_seat(**kwargs):
        stage = kwargs.get("execution_stage", "")
        agent_id = kwargs["agent_id"]
        turn_index = kwargs["turn_index"]
        calls.append((stage, agent_id, turn_index))
        stage_messages.setdefault(stage, []).append(str(kwargs.get("message") or ""))
        if stage == "manager_dispatch":
            return SeatResponse(agent_id=agent_id, turn_index=0, content=json.dumps(DISPATCH_PARALLEL), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage in ("expert", "rework"):
            seq = deliveries.get(agent_id) or [GOOD_DELIVERY]
            idx = delivery_idx.get(agent_id, 0)
            delivery_idx[agent_id] = idx + 1
            content = seq[min(idx, len(seq) - 1)]
            # The seat ran a real (fake) tool this turn: its successful result
            # becomes the trusted evidence ref the delivery may cite.
            content = content.replace("@TOOL_REF@", f"tool:kb_search:{turn_index}")
            return SeatResponse(
                agent_id=agent_id,
                turn_index=turn_index,
                content=content,
                tool_calls=[{"id": f"tc-{agent_id}-{turn_index}", "name": "kb_search"}],
                usage={"input_tokens": 1, "output_tokens": 1},
            )
        if stage == "manager_review":
            rnd = review_round[0]
            review_round[0] = rnd + 1
            plan = review_plan[rnd] if rnd < len(review_plan) else {}
            decisions = [
                {
                    "agent_id": agent,
                    "decision": d.get("decision", "accept"),
                    "reason": str(d.get("reason") or "r")[:1000],
                    "revision_requirements": list(d.get("requirements") or []),
                }
                for agent, d in plan.items()
            ]
            return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=json.dumps({"decisions": decisions}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "cross_review":
            content = (
                json.dumps({"summary": "ok", "evidence_ok": True, "confirmed_candidates": []})
                if cross_review_ok
                else json.dumps({"summary": "有缺口", "evidence_ok": False, "gaps": ["证据不足"], "confirmed_candidates": []})
            )
            return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=content, tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        if stage == "finalizer":
            return SeatResponse(agent_id=agent_id, turn_index=turn_index, content="final report", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
        return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=GOOD_DELIVERY, 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)
        result = {}
        for participant in participants:
            agent_id = participant["agent_id"]
            thread_id = f"thread-{agent_id}"
            await q_svc.store.participants.bind_thread(session_id, agent_id, thread_id=thread_id)
            result[agent_id] = thread_id
        return result

    monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat))
    monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(side_effect=_ensure_threads))
    if emit_real:
        real_emit_critical = runner._adapter._emit_critical
        monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock(side_effect=real_emit_critical))
    else:
        monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock())
    monkeypatch.setattr(runner._adapter, "_emit", AsyncMock())
    runner.stage_messages = stage_messages
    return runner, calls, ensure_threads_calls


async def _run(q_svc, monkeypatch, **kwargs):
    runner, calls, ensure_threads_calls = _make_q_runner(q_svc, monkeypatch, **kwargs)
    session = await q_svc.store.sessions.get_session("q-sess", user_id="u1")
    run = await q_svc.store.runs.get_run("q-run")
    await runner.execute(session, run)
    final_run = await q_svc.store.runs.get_run("q-run")
    final_session = await q_svc.store.sessions.get_session("q-sess", user_id="u1")
    return runner, calls, ensure_threads_calls, final_run, final_session


@pytest.mark.asyncio
async def test_runner_quality_accepts_good_deliveries(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    _, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [GOOD_DELIVERY], "seat-b": [GOOD_DELIVERY]},
        review_plan=[{"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}}],
    )

    assert final_run["status"] == "completed"
    assert final_session["status"] == "completed"

    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}
    assert set(by_key) == {"01:seat-a", "02:seat-b"}
    assert all(t["status"] == "accepted" for t in tasks)
    assert all(t["attempt_count"] == 1 for t in tasks)
    assert all(t["rework_count"] == 0 for t in tasks)

    checkpoint = final_run.get("maf_checkpoint") or {}
    assert checkpoint.get("seat_task_statuses", {}).get("01:seat-a", {}).get("status") == "accepted"
    assert checkpoint.get("controlled_gaps") == []

    result = final_session.get("final_result") or {}
    assert result.get("quality_decision") == "accept"
    assert result.get("controlled_gaps") == []

    review_stages = [s for s, a, t in calls if s == "manager_review"]
    assert len(review_stages) == 1, "one batched manager review round"


# ── 2. short answer → reworks → reassign → controlled gap ───────────────────


@pytest.mark.asyncio
async def test_runner_quality_rework_reassign_controlled_gap(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    # seat-a stays short across every attempt; seat-b is always good.
    # Manager keeps returning return_for_rework for seat-a (cap 2), then seat-a
    # is reassigned to seat-c whose delivery also fails → controlled gap.
    _, calls, ensure_threads_calls, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={
            "seat-a": [SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY],
            "seat-b": [GOOD_DELIVERY],
            "seat-c": [SHORT_DELIVERY],
        },
        review_plan=[
            {"seat-a": {"decision": "return_for_rework", "reason": "太短", "requirements": ["请补充证据"]}, "seat-b": {"decision": "accept"}},
            {"seat-a": {"decision": "return_for_rework", "reason": "仍太短", "requirements": ["请展开分析"]}},
            {"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
            {"seat-c": {"decision": "return_for_rework", "reason": "新席位也短"}},
        ],
    )

    # Run completes via the gap path (controlled gaps do NOT block the phase).
    assert final_run["status"] == "completed"
    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}

    task_a = by_key["01:seat-a"]
    assert task_a["status"] == "controlled_gap"
    assert task_a["attempt_count"] == 4          # seat-a 3 + seat-c 1
    assert task_a["rework_count"] == 2           # capped at max_rework_attempts
    assert task_a["reassign_count"] == 1         # exactly one reassignment
    assert task_a["agent_id"] == "seat-c"        # reassigned to backup
    assert task_a["original_agent_id"] == "seat-a"
    assert task_a["gap"] is not None
    assert "不得作为已验证事实" in task_a["gap"]["risk"]

    task_b = by_key["02:seat-b"]
    assert task_b["status"] == "accepted"
    assert task_b["attempt_count"] == 1

    # seat-c received a thread (backup handover) and delivered exactly once.
    assert "seat-c" in [aid for call in ensure_threads_calls for aid in call]
    seat_c_turns = [(s, a, t) for s, a, t in calls if a == "seat-c" and s in ("expert", "rework")]
    assert len(seat_c_turns) == 1

    # The quality gate passed with gaps and the final report carries them.
    checkpoint = final_run.get("maf_checkpoint") or {}
    gaps = checkpoint.get("controlled_gaps") or []
    assert len(gaps) == 1
    assert gaps[0]["task_key"] == "01:seat-a"

    result = final_session.get("final_result") or {}
    assert result.get("quality_decision") == "controlled_gap"
    assert len(result.get("controlled_gaps") or []) == 1
    assert result.get("degraded") is True

    # Manager review was batched once per quality round.
    review_stages = [t for s, a, t in calls if s == "manager_review"]
    assert len(review_stages) == 4


# ── 3. ask_user fallback pauses in waiting_input ────────────────────────────


@pytest.mark.asyncio
async def test_runner_quality_ask_user_waits_for_input(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    _, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
        review_plan=[
            {"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
            {"seat-a": {"decision": "return_for_rework", "reason": "仍太短"}},
            {"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
        ],
        fallback_policy="ask_user",
    )

    assert final_run["status"] == "waiting_input"
    assert final_session["status"] == "waiting_input"

    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}
    assert by_key["01:seat-a"]["status"] == "exhausted"
    assert by_key["02:seat-b"]["status"] == "accepted"

    # The finalizer must never run on the ask_user path.
    finalizer_stages = [s for s, a, t in calls if s == "finalizer"]
    assert finalizer_stages == []
    # No finished_at for a non-terminal pause.
    assert final_run.get("finished_at") is None


# ── 4. phase gate blocks on non-terminal required task ──────────────────────


@pytest.mark.asyncio
async def test_quality_gate_blocks_non_terminal_required_task(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    state = RoundtableState(
        session_id="q-sess", run_id="q-run", task_text="t",
        mode="structured", model_name=None, reasoning_effort=None,
        max_rounds=8, write_policy="auto", manager_agent_id=MANAGER,
        seat_agent_ids=["seat-a"], thread_map={}, capability_snapshot=None,
        seat_timeout_seconds=5, schema_repair_enabled=True,
        intent_type="parallel_consultation",
        dispatch_plan={"execution_mode": "parallel"},
        dispatch_contract_enabled=True, quality_gate_enabled=True,
    )
    state.seat_task_statuses = {
        "01:seat-a": {
            "task_key": "01:seat-a", "status": "submitted",
            "agent_id": "seat-a", "required": True,
        },
    }
    adapter = SimpleNamespace(
        _service=SimpleNamespace(store=SimpleNamespace(seat_tasks=SimpleNamespace())),
        _emit_critical=AsyncMock(),
    )
    gate = QualityGateExecutor(adapter, state)
    ctx = SimpleNamespace(send_message=AsyncMock(), yield_output=AsyncMock())

    with pytest.raises(RuntimeError, match="quality gate blocked"):
        await gate.handle({}, ctx)

    # The blocked gate must not masquerade as passed.
    assert state.status != "completed"


# ── 5. invalid/missing manager review NEVER auto-accepts ───────────────────


def _good_variant(n: int) -> str:
    """A deterministic-pass delivery whose conclusion differs per attempt, so
    duplicate detection never interferes with the missing-review behaviour."""
    return json.dumps({
        "summary": (
            f"结论{n}：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，"
            f"同时预留华南试点预算，第{n}次补充细化执行路径与里程碑安排"
        ),
        "findings": [f"依据{n}：华东区销售数据连续三个月增长", "依据二：竞品在华东投放低于均值"],
        "evidence_refs": ["@TOOL_REF@"],
        "open_questions": [],
        "self_check": "已覆盖全部验收标准",
    })


@pytest.mark.asyncio
async def test_runner_quality_missing_review_never_accepts(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    # The manager review returns garbage (empty decisions) every round. Even
    # though every delivery passes the deterministic length/format/evidence
    # checks, NO task may be accepted without a valid manager review: the
    # server forces system reworks, then exhausts into controlled gaps.
    variants_a = [_good_variant(i) for i in range(1, 7)]
    variants_b = [_good_variant(i + 10) for i in range(1, 7)]
    _, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={
            "seat-a": variants_a + [_good_variant(90)],
            "seat-b": variants_b + [_good_variant(91)],
            "seat-c": [_good_variant(80)],
        },
        review_plan=[{}],
        emit_real=True,
    )

    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    assert tasks, "seat tasks must exist"
    for task in tasks:
        assert task["status"] != "accepted", (
            f"task {task['task_key']} was accepted without a valid manager review"
        )
        assert task["status"] == "controlled_gap"
        assert task["rework_count"] == 2
        assert task["reassign_count"] == 1
        decision = task.get("review_decision") or {}
        assert decision.get("source") == "system"
        assert decision.get("reason") == "manager_review_missing_or_invalid"

    # The run still terminates (gaps do not block the phase) — but only as a
    # degraded controlled-gap result, never as a clean accept.
    assert final_run["status"] == "completed"
    result = final_session.get("final_result") or {}
    assert result.get("quality_decision") == "controlled_gap"
    assert len(result.get("controlled_gaps") or []) == len(tasks)

    # The persisted review events carry source=system + the safe reason.
    events = await q_svc.store.events.list_events(run_id="q-run")
    review_events = [e for e in events if e["event_type"] == "manager.review.decision"]
    assert review_events
    system_events = [e for e in review_events if e["payload"].get("source") == "system"]
    assert system_events
    assert all(
        e["payload"].get("reason") == "manager_review_missing_or_invalid"
        for e in system_events
    )


# ── 6. unaccepted deliveries never reach facts / review / final report ─────


PLAUSIBLE_GAP_DELIVERY = json.dumps({
    # Fails below_min_length on purpose, but carries "plausible" findings and
    # an invented evidence ref — none of it may survive into the ledger, the
    # cross-review prompt or the final synthesis.
    "summary": "好",
    "findings": ["PLAUSIBLE_FINDING: 已核实华东区增长数据并确认竞品全面撤离"],
    "evidence_refs": ["plausible-ref-001"],
    "open_questions": [],
    "self_check": "ok",
})


@pytest.mark.asyncio
async def test_runner_quality_gap_content_isolated_from_facts_and_report(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    runner, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={
            "seat-a": [PLAUSIBLE_GAP_DELIVERY],
            "seat-b": [GOOD_DELIVERY],
            "seat-c": [PLAUSIBLE_GAP_DELIVERY],
        },
        review_plan=[
            {"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
            {"seat-a": {"decision": "return_for_rework", "reason": "仍太短"}},
            {"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
            {"seat-c": {"decision": "return_for_rework", "reason": "新席位也不合格"}},
        ],
    )

    assert final_run["status"] == "completed"
    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}
    assert by_key["01:seat-a"]["status"] == "controlled_gap"
    assert by_key["02:seat-b"]["status"] == "accepted"

    # 1) Fact ledger: the gapped seat's claims/refs are absent; the accepted
    #    seat's findings ARE present (merge happens on acceptance).
    ledger = final_session.get("fact_ledger") or {}
    candidate_texts = [
        str(c.get("text") or "") for c in ledger.get("candidate_facts") or []
    ]
    confirmed_texts = [
        str(f.get("text") or "") for f in ledger.get("confirmed_facts") or []
    ]
    evidence_texts = [
        str(e.get("ref") or e.get("text") or "") for e in ledger.get("evidence") or []
    ]
    assert not any("PLAUSIBLE_FINDING" in t for t in candidate_texts)
    assert not any("PLAUSIBLE_FINDING" in t for t in confirmed_texts)
    assert not any("plausible-ref-001" in t for t in evidence_texts)
    assert any("华东区销售数据连续三个月增长" in t for t in candidate_texts)

    # 2) Cross-review prompt: accepted seat only, never the gapped content.
    cross_review_msgs = runner.stage_messages.get("cross_review") or []
    assert cross_review_msgs
    assert all("PLAUSIBLE_FINDING" not in m for m in cross_review_msgs)
    assert all("plausible-ref-001" not in m for m in cross_review_msgs)
    assert any("华东区销售数据连续三个月增长" in m for m in cross_review_msgs)

    # 3) Final synthesis: accepted seat only + traceable gap risk section.
    finalizer_msgs = runner.stage_messages.get("finalizer") or []
    assert finalizer_msgs
    assert all("PLAUSIBLE_FINDING" not in m for m in finalizer_msgs)
    assert all("plausible-ref-001" not in m for m in finalizer_msgs)
    assert any("01:seat-a" in m for m in finalizer_msgs), "gap must be traceable"
    assert any("受控缺口" in m for m in finalizer_msgs)

    # 4) Checkpoint: exactly one accepted result (the accepted seat).
    checkpoint = final_run.get("maf_checkpoint") or {}
    accepted_results = checkpoint.get("accepted_task_results") or []
    assert len(accepted_results) == 1
    assert accepted_results[0]["task_key"] == "02:seat-b"
    assert accepted_results[0]["accepted"] is True
    assert "PLAUSIBLE_FINDING" not in str(accepted_results)

    # 5) Final result carries the gap as risk, never as a verified fact.
    result = final_session.get("final_result") or {}
    assert result.get("quality_decision") == "controlled_gap"
    gaps = result.get("controlled_gaps") or []
    assert len(gaps) == 1
    assert gaps[0]["task_key"] == "01:seat-a"
    assert "不得作为已验证事实" in gaps[0]["risk"]


@pytest.mark.asyncio
async def test_runner_quality_events_persisted_and_projected(q_svc, monkeypatch):
    """The Stage-4 audit events persist to maf_roundtable_events."""
    pytest.importorskip("agent_framework")

    _, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
        review_plan=[{"seat-a": {"decision": "return_for_rework", "requirements": ["补证据"]}, "seat-b": {"decision": "accept"}}],
        emit_real=True,
    )

    events = await q_svc.store.events.list_events(run_id="q-run")
    types = {e["event_type"] for e in events}
    assert "seat.delivery.submitted" in types
    assert "manager.review.decision" in types
    assert "seat.rework.assigned" in types
    assert "stage.quality_gate" in types

    # The short delivery was flagged deterministically in the persisted event.
    deliveries = [e for e in events if e["event_type"] == "seat.delivery.submitted"]
    seat_a = [e for e in deliveries if e.get("agent_id") == "seat-a"]
    assert seat_a and seat_a[0]["payload"].get("deterministic_passed") is False


# ── 7. P1: zero config values are honoured, not coerced to defaults ────────


def test_quality_settings_honour_zero_config_values(monkeypatch):
    from app.maf_roundtable.workflow_runner import WorkflowRunner

    fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(
        quality_gate_enabled=True,
        max_rework_attempts=0,
        max_reassignments=0,
        min_delivery_chars=0,
        min_delivery_evidence=0,
        default_fallback_policy="controlled_gap",
    ))
    monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)

    runner = WorkflowRunner.__new__(WorkflowRunner)  # no __init__ side effects
    settings = runner._quality_settings()
    assert settings["max_rework_attempts"] == 0
    assert settings["max_reassignments"] == 0
    assert settings["min_delivery_chars"] == 0
    assert settings["min_delivery_evidence"] == 0
    assert settings["fallback_policy"] == "controlled_gap"


def test_quality_settings_none_falls_back_to_defaults(monkeypatch):
    from app.maf_roundtable.workflow_runner import WorkflowRunner

    # Attributes absent entirely (getattr → None) must fall back to defaults.
    fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(
        quality_gate_enabled=True,
    ))
    monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)

    runner = WorkflowRunner.__new__(WorkflowRunner)
    settings = runner._quality_settings()
    assert settings["max_rework_attempts"] == 2
    assert settings["max_reassignments"] == 1
    assert settings["min_delivery_chars"] == 40
    assert settings["min_delivery_evidence"] == 1
    assert settings["fallback_policy"] == "reassign_then_gap"


@pytest.mark.asyncio
async def test_runner_quality_zero_rework_cap_no_rework_granted(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    # max_rework_attempts=0 must be honoured: a failing delivery is NOT sent
    # back for rework — it exhausts straight into the fallback (controlled gap
    # here, no backup reassignment budget either).
    _, calls, _, final_run, final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
        review_plan=[
            {"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
        ],
        config_overrides={"max_rework_attempts": 0, "max_reassignments": 0},
    )

    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}
    task_a = by_key["01:seat-a"]
    assert task_a["status"] == "controlled_gap"
    assert task_a["attempt_count"] == 1, "no rework delivery may happen with a 0 cap"
    assert task_a["rework_count"] == 0
    assert task_a["reassign_count"] == 0
    # Only the initial delivery + the (ignored) rework request — no rework turn.
    expert_turns_a = [s for s, a, _t in calls if a == "seat-a" and s in ("expert", "rework")]
    assert len(expert_turns_a) == 1

    result = final_session.get("final_result") or {}
    assert result.get("quality_decision") == "controlled_gap"


@pytest.mark.asyncio
async def test_runner_quality_zero_min_evidence_accepts_refless_delivery(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    # min_delivery_evidence=0 must be honoured: a conclusion-only delivery with
    # no evidence refs passes the deterministic evidence gate.
    no_ref_delivery = json.dumps({
        "summary": "结论：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，同时预留华南试点预算",
        "findings": [],
        "evidence_refs": [],
        "open_questions": [],
        "self_check": "ok",
    })
    _, _calls, _, final_run, _final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [no_ref_delivery], "seat-b": [no_ref_delivery]},
        review_plan=[{"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}}],
        config_overrides={"min_delivery_evidence": 0},
    )

    assert final_run["status"] == "completed"
    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    assert all(t["status"] == "accepted" for t in tasks)


# ── 8. P1: restart recovery re-submission hits duplicate via digest ────────


@pytest.mark.asyncio
async def test_runner_quality_restart_resubmission_hits_duplicate_digest(q_svc, monkeypatch):
    pytest.importorskip("agent_framework")

    from app.maf_roundtable.quality_contracts import build_task_contract, conclusion_digest

    good = json.loads(GOOD_DELIVERY.replace("@TOOL_REF@", "tool:kb_search:1"))
    digest = conclusion_digest(good["summary"])

    # Simulate the pre-restart state: seat-a's first delivery was rejected and
    # its conclusion digest was persisted to the task history; then the process
    # restarted. The task row (with history) survives in the DB.
    contract = build_task_contract(
        task_goal="市场增长机会分析", delivery_requirements=["证据", "优先级"],
        max_rework_attempts=2, fallback_policy="reassign_then_gap",
    )
    seeded = await q_svc.store.seat_tasks.create_task(
        session_id="q-sess", run_id="q-run", task_key="01:seat-a",
        agent_id="seat-a", contract=contract.model_dump(), required=True,
    )
    await q_svc.store.seat_tasks.update_task(seeded["id"], {
        "status": "rework",
        "attempt_count": 1,
        "rework_count": 1,
        "history": [{
            "attempt": 1, "agent_id": "seat-a", "round": 1,
            "schema_status": "ok", "failure_codes": ["below_min_length"],
            "conclusion_sha256": digest,
        }],
    })

    # After the restart the seat re-submits the SAME conclusion. It must be
    # flagged duplicate_content from the persisted digest — even though the
    # in-memory prior_conclusions starts empty on a fresh process.
    _, _calls, _, final_run, _final_session = await _run(
        q_svc, monkeypatch,
        deliveries={"seat-a": [GOOD_DELIVERY], "seat-b": [GOOD_DELIVERY]},
        review_plan=[
            {"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}},
            {"seat-a": {"decision": "accept"}},
            {"seat-a": {"decision": "accept"}},
            {"seat-a": {"decision": "accept"}},
        ],
    )

    tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
    by_key = {t["task_key"]: t for t in tasks}
    task_a = by_key["01:seat-a"]
    # The re-submission was deterministically flagged as a duplicate.
    check = task_a.get("deterministic_check") or {}
    history_codes = [
        code
        for h in (task_a.get("history") or [])
        for code in (h.get("failure_codes") or [])
    ]
    assert "duplicate_content" in (check.get("failure_codes") or []) or "duplicate_content" in history_codes
    # And a duplicate can never be accepted, even with the manager saying yes.
    assert task_a["status"] != "accepted"
    assert final_run["status"] == "completed"