"""Phase 2 P1 acceptance tests for the MAF roundtable workflow execution.

These tests exercise the REAL MAF SDK workflow orchestration (WorkflowBuilder,
Executor subclasses, streaming events, checkpoint persistence) with a mocked
DeerFlow runtime client (no real LLM calls). They verify the four completion
criteria for Phase 2 P1 closure:

1. Real MAF Workflow is running.
2. Write tools never execute before approval (interception before execution).
3. After approval, the workflow resumes and executes exactly once (at-most-once).
4. Cancel and restart recovery don't leave DeerFlow orphan tasks.

Additionally covers:
- Two seats each create DeerFlow threads owned by the session user.
- MAF group_chat and structured both run a real SDK path.
- Secrets (token/cookie/password/long secret) never appear in events/errors/logs.
- source_event_id dedup at DB level.
- Audit event write failure →external write does not execute (run fails).
- ActionCoordinator: encrypted params never returned from API/events/logs.
"""

from __future__ import annotations

import asyncio
import itertools
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

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

from app.maf_roundtable.action_coordinator import (
    ActionCoordinator,
    classify_tool,
    decrypt_params,
    encrypt_params,
    tool_needs_approval,
)
from app.maf_roundtable.deerflow_agent_adapter import DeerFlowAgentAdapter
from app.maf_roundtable.runtime_client import ExecutionPrincipal
from app.maf_roundtable.workflow_runner import WorkflowRunner
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store

# ── fixtures ──────────────────────────────────────────────────────────────


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

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


def _make_app(store):
    """Build a minimal fake app with the state the runner needs."""
    app = SimpleNamespace()
    app.state = SimpleNamespace()
    app.state.maf_roundtable_service = _svc(store)
    app.state.checkpointer = AsyncMock()
    app.state.thread_store = AsyncMock()
    app.state.run_manager = AsyncMock()
    app.state.stream_bridge = None
    return app


class _FakeRunRecord:
    """Mimics the RunRecord returned by start_run."""

    def __init__(self, run_id="df-run-1"):
        self.run_id = run_id
        self.task = asyncio.ensure_future(asyncio.sleep(0))


def _patch_runtime_client(adapter: DeerFlowAgentAdapter, *, responses: dict[str, str] | None = None, tool_calls_map: dict[str, list] | None = None):
    """Patch the adapter's runtime client to return controlled responses.

    ``responses`` maps agent_id →content text.
    ``tool_calls_map`` maps agent_id →list of tool call dicts the AI "generated".
    """
    responses = responses or {}
    tool_calls_map = tool_calls_map or {}
    call_log: list[dict] = []
    write_tool_counter: dict[str, int] = {"count": 0}

    async def fake_create_thread(principal: ExecutionPrincipal) -> str:
        return f"thread-{principal.session_id[:8]}-{len(call_log)}"

    async def fake_start_agent_run(principal, *, thread_id, agent_id, message, excluded_tools=None, approved_tool=None, turn_index=0, capability_snapshot=None, write_policy="confirm_high_risk", **kwargs):
        call_log.append({
            "agent_id": agent_id,
            "thread_id": thread_id,
            "message": message[:200],
            "excluded_tools": list(excluded_tools or []),
            "approved_tool": approved_tool,
            "turn_index": turn_index,
            "capability_snapshot": capability_snapshot,
            "write_policy": write_policy,
        })

        # Phase-2 simulation: when this is a normal seat turn (no approved_tool),
        # the WriteGuard middleware would intercept write tools inside the real
        # DeerFlow run and create pending actions. Since the fake bypasses the
        # middleware, we replay that here so the action/approval get created
        # exactly as the real middleware would.
        if not approved_tool and write_policy != "auto":
            from app.maf_roundtable.action_coordinator import ActionCoordinator

            tcs = tool_calls_map.get(agent_id, [])
            coord = ActionCoordinator(adapter._service, user_id=principal.user_id)
            try:
                await coord.intercept_write_tool_calls(
                    run_id=principal.run_id,
                    session_id=principal.session_id,
                    agent_id=agent_id,
                    turn_index=turn_index,
                    tool_calls=tcs,
                    capability_snapshot=capability_snapshot,
                    write_policy=write_policy,
                )
            except Exception:
                pass  # best-effort replay for the test harness

        return _FakeRunRecord(run_id=f"df-{agent_id}")

    async def fake_read_messages(principal, thread_id):
        agent_id = ""
        for c in call_log:
            if c["thread_id"] == thread_id:
                agent_id = c["agent_id"]
                break

        # Phase-2 semantics: the WriteGuard middleware intercepts write tools
        # inside the DeerFlow run. A normal seat turn (no ``approved_tool``)
        # means the write was intercepted →counter stays put. Only the
        # ToolExecutor's deterministic run (via execute_tool_directly) sets
        # the write counter.
        tcs = tool_calls_map.get(agent_id, [])
        content = responses.get(agent_id, f"Response from {agent_id}")
        messages = [{"type": "ai", "content": content, "usage_metadata": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}}]
        if tcs:
            messages[0]["tool_calls"] = tcs
        return messages

    async def fake_execute_tool_directly(*, thread_id, agent_id, tool_name, params, principal, **kwargs):
        """Simulate direct tool execution —the LLM is bypassed (P0-2).
        Increment the write counter for the approved tool. Extra P0-1 context
        kwargs (turn_index/capability_snapshot/— are accepted and ignored."""
        write_tool_counter["count"] += 1
        return {"status": "succeeded", "result": {"tool": tool_name, "params": params}}

    adapter._client.create_thread = fake_create_thread
    adapter._client.start_agent_run = fake_start_agent_run
    adapter._client.read_thread_messages = fake_read_messages
    adapter._client.cancel_run = AsyncMock(return_value=True)
    adapter._client.execute_tool_directly = fake_execute_tool_directly
    return call_log, write_tool_counter


# ── 1. Real SDK structured workflow, 2 seats ─────────────────────────────


@pytest.mark.asyncio
async def test_real_sdk_structured_workflow_two_seats(store):
    """Build and run a real MAF SDK structured workflow with 2 seat executors."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "分析市场趋势",
            "mode": "structured",
            "participants": [
                {"agent_id": "analyst-a", "role": "seat"},
                {"agent_id": "analyst-b", "role": "seat"},
            ],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(runner.adapter, responses={
        "analyst-a": "市场呈上升趋势",
        "analyst-b": "需注意风险因素",
    })

    await runner.execute(s, run)

    final_run = await svc.get_run(run_id=run["id"], user_id="u1")
    assert final_run["status"] == "completed"
    assert final_run["finished_at"] is not None
    assert final_run["started_at"] is not None

    final_session = await svc.get_session(session_id=s["id"], user_id="u1")
    assert final_session["status"] == "completed"
    assert final_session["final_result"] is not None


# ── 2. Two seats create threads owned by session user ──────────────────


@pytest.mark.asyncio
async def test_two_seats_create_threads_owned_by_session_user(store):
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "participants": [
                {"agent_id": "seat-1", "role": "seat"},
                {"agent_id": "seat-2", "role": "seat"},
            ],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")

    created_threads: list[dict] = []
    orig_create = runner.adapter._client.create_thread

    async def tracking_create(principal: ExecutionPrincipal):
        tid = await orig_create(principal)
        created_threads.append({"thread_id": tid, "user_id": principal.user_id, "session_id": principal.session_id})
        return tid

    runner.adapter._client.create_thread = tracking_create
    _patch_runtime_client(runner.adapter)
    runner.adapter._client.create_thread = tracking_create

    await runner.execute(s, run)

    assert len(created_threads) == 2
    assert all(t["user_id"] == "u1" for t in created_threads)
    assert all(t["session_id"] == s["id"] for t in created_threads)
    assert created_threads[0]["thread_id"] != created_threads[1]["thread_id"]


# ── 3. Two seats execute in order, producing events ─────────────────────


@pytest.mark.asyncio
async def test_two_seats_execute_in_order_producing_events(store):
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "participants": [
                {"agent_id": "seat-x", "role": "seat"},
                {"agent_id": "seat-y", "role": "seat"},
            ],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(runner.adapter)

    await runner.execute(s, run)

    events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
    event_types = [e["event_type"] for e in events]

    assert event_types.count("agent.turn_started") >= 2
    assert event_types.count("agent.message.completed") >= 2
    assert "run.completed" in event_types


# ── 4. Seat tool call + skill + artifact events ─────────────────────────


@pytest.mark.asyncio
async def test_seat_tool_call_and_skill_artifact_events(store):
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "participants": [{"agent_id": "tool-seat", "role": "seat"}],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")

    from app.maf_roundtable.event_normalizer import (
        make_artifact_created,
        make_skill_completed,
        make_skill_started,
    )

    await runner.adapter._emit_critical(run["id"], {
        "event_type": "tool.requested",
        "agent_id": "tool-seat",
        "turn_index": 1,
        "payload": {"tool_name": "web_search", "tool_call_id": "tc1"},
    })
    await runner.adapter._emit_critical(run["id"], {
        "event_type": "tool.completed",
        "agent_id": "tool-seat",
        "turn_index": 1,
        "payload": {"tool_name": "web_search", "tool_call_id": "tc1", "status": "completed"},
    })
    await runner.adapter._emit_critical(run["id"], make_skill_started("tool-seat", 1, skill_name="data-analysis"))
    await runner.adapter._emit_critical(run["id"], make_skill_completed("tool-seat", 1, skill_name="data-analysis", result_preview="分析完成"))
    await runner.adapter._emit_critical(run["id"], make_artifact_created("tool-seat", 1, artifact_type="chart", artifact_id="art-1", title="趋势图"))

    events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
    types = [e["event_type"] for e in events]
    assert "tool.requested" in types
    assert "tool.completed" in types
    assert "skill.started" in types
    assert "skill.completed" in types
    assert "artifact.created" in types


# ── 5. P1: Write tool counter —0 before approval, 1 after, still 1 after repeated resume ──


@pytest.mark.asyncio
async def test_write_tool_counter_zero_before_approval(store):
    """P1: write tools are intercepted before execution. The write tool
    counter must be 0 before any approval."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "分析完成，需要创建工单"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)

    # Write tool was excluded →never executed →counter is 0.
    assert write_counter["count"] == 0, "write tool executed before approval!"

    # An action + approval should have been created.
    actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    assert len(actions) == 1
    assert actions[0]["tool_name"] == "create_ticket"
    assert actions[0]["status"] == "waiting_approval"

    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    assert len(approvals) == 1
    assert approvals[0]["status"] == "pending"

    # Run should be in waiting_approval (not completed).
    final_run = await svc.get_run(run_id=run["id"], user_id="u1")
    assert final_run["status"] == "waiting_approval"
    # waiting_approval must NOT write finished_at.
    assert final_run["finished_at"] is None


@pytest.mark.asyncio
async def test_write_tool_counter_one_after_approval(store):
    """P1: after approval, the workflow resumes and the write tool executes
    exactly once."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    # Execute the workflow →creates action + approval →waiting_approval.
    await runner.execute(s, run)

    assert write_counter["count"] == 0  # Before approval: 0.

    # Approve.
    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    approval = approvals[0]
    decided = await svc.decide_approval(approval_id=approval["id"], decision="approved", user_id="u1")
    assert decided["status"] == "approved"

    # Resume the workflow.
    resumed = await runner.resume(run["id"])
    assert resumed is not None
    assert resumed["status"] == "completed"

    # After approval + resume: write tool executed exactly once.
    assert write_counter["count"] == 1, "write tool should execute exactly once after approval"

    # Action should be succeeded.
    actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    assert actions[0]["status"] == "succeeded"


@pytest.mark.asyncio
async def test_write_tool_still_one_after_repeated_resume(store):
    """P1: repeated resume does not double-execute the write tool."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)

    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")

    # First resume.
    await runner.resume(run["id"])
    assert write_counter["count"] == 1

    # Second resume —should not re-execute.
    await runner.resume(run["id"])
    assert write_counter["count"] == 1, "repeated resume must not double-execute"


@pytest.mark.asyncio
async def test_concurrent_resume_second_sees_running_tool_runs_once(store):
    """P0-3: two concurrent resumes. The first wins the run-level CAS and blocks
    inside the tool; the second sees the run already ``running`` and returns
    WITHOUT writing a terminal status. The tool executes exactly once; only after
    the first resume's tool is released does the run finalize (report) →completed.
    """
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    # Replace the direct-tool executor with a blocking one so we can interleave
    # a second resume while the first is mid-execution.
    write_counter = {"count": 0}
    started = asyncio.Event()
    release = asyncio.Event()

    async def blocking_execute_tool_directly(*, thread_id, agent_id, tool_name, params, principal, **kwargs):
        write_counter["count"] += 1
        started.set()          # signal: first resume has claimed run+action and is executing
        await release.wait()   # block until the test releases it
        return {"status": "succeeded", "result": {"tool": tool_name}}

    runner.adapter._client.execute_tool_directly = blocking_execute_tool_directly

    # Drive the workflow to waiting_approval, then approve.
    await runner.execute(s, run)
    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")

    # First resume (background) —wins the run CAS, then blocks in the tool.
    first_task = asyncio.ensure_future(runner.resume(run["id"]))
    await asyncio.wait_for(started.wait(), timeout=5)
    assert write_counter["count"] == 1, "tool must have started exactly once"

    # Second resume —the run is already 'running'; it must return that and must
    # NOT write a terminal status or re-execute the tool.
    second = await runner.resume(run["id"])
    assert second is not None and second["status"] == "running", \
        f"second concurrent resume must see running, got {second and second['status']}"
    assert write_counter["count"] == 1, "second resume must not re-execute the tool"

    # Release the first resume's tool →it finalizes (report) →completed.
    release.set()
    first = await asyncio.wait_for(first_task, timeout=10)
    assert first is not None and first["status"] == "completed"
    assert write_counter["count"] == 1, "tool executes exactly once across both resumes"

    # The completed run carries a synthesized final report (finalizer re-entered).
    final_run = await svc.store.runs.get_run(run["id"])
    assert final_run["finished_at"] is not None


@pytest.mark.asyncio
async def test_resume_with_missing_secret_fails_without_executing(store):
    """P0-2: if the action secret is deleted before resume, the write tool must
    execute **0 times**, the action is ``failed`` (not succeeded), and the parent
    run is ``failed`` —never completed, never a retry of the external write."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)
    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")

    # Delete the encrypted secret out from under the approved action.
    actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    action_id = actions[0]["id"]
    await store.action_secrets.delete_by_action(action_id)

    resumed = await runner.resume(run["id"])

    assert write_counter["count"] == 0, "tool must NOT execute when the secret is gone"
    final_actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    assert final_actions[0]["status"] == "failed"
    assert final_actions[0]["status"] != "succeeded"
    assert resumed is not None and resumed["status"] == "failed"


@pytest.mark.asyncio
async def test_resume_with_corrupt_secret_fails_without_executing(store):
    """P0-2: if the action secret cannot be decrypted (wrong key / garbage),
    resume fails the action and the run with the tool executing **0 times**."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)
    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")

    # Replace the ciphertext with garbage that Fernet cannot decrypt.
    actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    action_id = actions[0]["id"]
    await store.action_secrets.delete_by_action(action_id)
    await store.action_secrets.create(action_id=action_id, arguments_encrypted="corrupt-token")

    resumed = await runner.resume(run["id"])

    assert write_counter["count"] == 0, "tool must NOT execute when the secret is undecryptable"
    final_actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    assert final_actions[0]["status"] == "failed"
    assert resumed is not None and resumed["status"] == "failed"


# ── 6. P1: Rejected approval →write count stays 0 ──────────────────────


@pytest.mark.asyncio
async def test_rejected_approval_write_count_stays_zero(store):
    """P1: after a rejected approval, the external write count stays 0 and
    the action is marked rejected."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _, write_counter = _patch_runtime_client(
        runner.adapter,
        responses={"writer": "需要创建工单"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)

    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="rejected", user_id="u1")

    await runner.resume(run["id"])

    assert write_counter["count"] == 0, "rejected approval must not execute write tool"

    actions = await svc.list_actions(run_id=run["id"], user_id="u1")
    assert actions[0]["status"] == "rejected"


# ── 7. P1: In-process cancel cascades to DeerFlow ───────────────────────


@pytest.mark.asyncio
async def test_in_process_cancel_cascades_to_deerflow(store):
    """P1: cancel calls DeerFlow run_manager.cancel on all active seat runs."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={"task_text": "t", "participants": [{"agent_id": "seat-1", "role": "seat"}]},
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    # Create a seat run with a DeerFlow run id.
    seat_run = await store.seat_runs.create_seat_run(
        maf_run_id=run["id"], agent_id="seat-1", turn_index=1, deerflow_thread_id="thread-x"
    )
    await store.seat_runs.mark_running(seat_run["id"], deerflow_run_id="df-run-42")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    runner.adapter._client.cancel_run = AsyncMock(return_value=True)

    result = await runner.cancel(run["id"])
    assert result["all_cancelled"] is True

    # DeerFlow cancel_run was called with the right run id.
    runner.adapter._client.cancel_run.assert_called_once_with("df-run-42")

    # Seat run should be cancelled.
    seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
    assert seat_runs[0]["status"] == "cancelled"


# ── 8. P1: Simulated restart cancel via persisted seat runs ──────────────


@pytest.mark.asyncio
async def test_restart_cancel_via_persisted_seat_runs(store):
    """P1: after a simulated restart, the system can still cancel outstanding
    DeerFlow runs via persisted seat_run records."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={"task_text": "t", "participants": [{"agent_id": "long-seat", "role": "seat"}]},
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    seat_run = await store.seat_runs.create_seat_run(
        maf_run_id=run["id"], agent_id="long-seat", turn_index=1, deerflow_thread_id="thread-x"
    )
    await store.seat_runs.mark_running(seat_run["id"], deerflow_run_id="df-run-99")

    # "Restart" —build a fresh runner (new adapter, new client).
    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    runner.adapter._client.cancel_run = AsyncMock(return_value=True)

    cancelled = await runner.cancel(run["id"])
    assert cancelled["all_cancelled"] is True
    runner.adapter._client.cancel_run.assert_called_once_with("df-run-99")

    seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
    assert seat_runs[0]["status"] == "cancelled"


# ── 9. P1: Two-phase cancel —cancelling then cancelled ─────────────────


@pytest.mark.asyncio
async def test_two_phase_cancel_cancelling_then_cancelled(store):
    """P1: cancel first marks seat runs as cancelling, then cancelled after
    DeerFlow cancel succeeds."""
    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")

    sr = await store.seat_runs.create_seat_run(
        maf_run_id=run["id"], agent_id="s1", turn_index=1, deerflow_thread_id="t1"
    )
    await store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")

    infos = await store.seat_runs.cancel_all_active(run["id"])
    assert len(infos) == 1
    assert infos[0]["deerflow_run_id"] == "df-1"

    runs = await store.seat_runs.get_by_maf_run(run["id"])
    assert runs[0]["status"] == "cancelling"

    await store.seat_runs.mark_cancelled(sr["id"])
    runs = await store.seat_runs.get_by_maf_run(run["id"])
    assert runs[0]["status"] == "cancelled"


@pytest.mark.asyncio
async def test_cancel_failed_when_deerflow_cancel_fails(store):
    """P1: when DeerFlow cancel fails, the seat run is marked cancel_failed."""
    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")

    sr = await store.seat_runs.create_seat_run(
        maf_run_id=run["id"], agent_id="s1", turn_index=1, deerflow_thread_id="t1"
    )
    await store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    runner.adapter._client.cancel_run = AsyncMock(return_value=False)

    result = await runner.cancel(run["id"])
    assert result["all_cancelled"] is False

    runs = await store.seat_runs.get_by_maf_run(run["id"])
    assert runs[0]["status"] == "cancel_failed"


# ── 10. MAF group_chat runs a real SDK path ──────────────────────────────


@pytest.mark.asyncio
async def test_group_chat_runs_real_sdk_path(store):
    """P1: group_chat mode runs a real MAF SDK workflow path."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "讨论方案",
            "mode": "group_chat",
            "max_rounds": 2,
            "participants": [
                {"agent_id": "g-seat-1", "role": "seat"},
                {"agent_id": "g-seat-2", "role": "seat"},
            ],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(runner.adapter, responses={
        "g-seat-1": "我认为方案A更好",
        "g-seat-2": "我同意",
    })

    await runner.execute(s, run)

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


# ── 11. Audit event failure prevents external write ──────────────────────


@pytest.mark.asyncio
async def test_audit_event_failure_prevents_external_write(store):
    """P1: when a critical audit event fails to persist, the adapter raises
    AuditEventError and the run fails —external writes never execute."""
    from app.maf_roundtable.deerflow_agent_adapter import AuditEventError

    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")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")

    with patch.object(svc, "append_event", side_effect=RuntimeError("DB write failed")):
        with pytest.raises(AuditEventError, match="failed to persist"):
            await runner.adapter._emit_critical(run["id"], {
                "event_type": "tool.requested",
                "agent_id": "seat",
                "turn_index": 1,
                "payload": {"tool_name": "write_data"},
            })


# ── 12. DeerFlow task error marks run failed ─────────────────────────────


@pytest.mark.asyncio
async def test_deerflow_task_error_marks_run_failed(store):
    """P3: when a DeerFlow seat turn raises, the seat is soft-failed
    (``partial_failure``) and the structured run still completes degraded;
    secrets must not appear in persisted seat / run errors."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={"task_text": "t", "participants": [{"agent_id": "boom-seat", "role": "seat"}]},
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")

    async def exploding_start(*args, **kwargs):
        raise RuntimeError("LLM provider unavailable: token=sk-secret-12345")

    runner.adapter._client.create_thread = AsyncMock(return_value="thread-1")
    runner.adapter._client.start_agent_run = exploding_start

    await runner.execute(s, run)

    final_run = await svc.get_run(run_id=run["id"], user_id="u1")
    assert final_run["status"] == "completed"
    assert final_run["finished_at"] is not None
    ckpt = final_run.get("maf_checkpoint") or {}
    assert ckpt.get("partial_failure") is True
    outcome = (ckpt.get("seat_outcomes") or {}).get("boom-seat") or {}
    assert outcome.get("status") in ("failed", "error")
    persisted = " ".join(
        [
            str(final_run.get("error_message_safe") or ""),
            str(outcome.get("error") or ""),
            str(ckpt),
        ]
    )
    assert "sk-secret-12345" not in persisted


# ── 13. Secrets never in events / errors / logs ──────────────────────────


@pytest.mark.asyncio
async def test_secrets_never_in_events_or_errors(store):
    """P1: tokens, cookies, passwords, and long secrets are stripped from
    event payloads, error messages, and seat-run errors."""
    from app.maf_roundtable.sanitize import sanitize

    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")

    secret_payload = {
        "tool_name": "api_call",
        "authorization": "Bearer sk-live-abcdef123456789",
        "cookie": "session=abc123; token=xyz789",
        "password": "hunter2",
        "api_key": "sk-proj-1234567890abcdef",
        "nested": {"secret": "deep-secret-value", "data": "safe"},
        "long_text": "x" * 600 + " token=embedded-secret-999 " + "y" * 600,
    }
    event = await svc.append_event(
        run_id=run["id"], user_id="u1", event_type="tool.completed", payload=secret_payload
    )

    stored = event["payload"]
    assert stored["authorization"] == "***REDACTED***"
    assert stored["cookie"] == "***REDACTED***"
    assert stored["password"] == "***REDACTED***"
    assert stored["api_key"] == "***REDACTED***"
    assert stored["nested"]["secret"] == "***REDACTED***"
    assert stored["nested"]["data"] == "safe"
    assert "embedded-secret-999" not in stored["long_text"]

    long_secret = "prefix " * 200 + "password=super-secret-value " + "suffix " * 200
    sanitized = sanitize(long_secret)
    assert "super-secret-value" not in sanitized


# ── 14. source_event_id dedup at DB level ────────────────────────────────


@pytest.mark.asyncio
async def test_source_event_id_dedup_at_db_level(store):
    """P1: events with the same (run_id, source_event_id) are deduped at the DB level."""
    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")

    e1 = await svc.append_event(
        run_id=run["id"], user_id="u1", event_type="agent.message.delta",
        payload={"text": "hello"}, source_event_id="msg-abc-123",
    )
    e2 = await svc.append_event(
        run_id=run["id"], user_id="u1", event_type="agent.message.delta",
        payload={"text": "hello again"}, source_event_id="msg-abc-123",
    )

    assert e1["id"] == e2["id"]
    assert e1["seq"] == e2["seq"]

    events, latest = await svc.list_events(run_id=run["id"], user_id="u1")
    assert len(events) == 1
    assert latest == 1


# ── 15. ActionCoordinator: encrypted params never returned ───────────────


@pytest.mark.asyncio
async def test_encrypted_action_params_never_returned(store):
    """P1: encrypted actual execution params are never returned from the
    actions API, events, or logs."""
    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")

    coord = ActionCoordinator(svc, user_id="u1")

    # Create an action with encrypted params.
    action = await store.actions.create_action({
        "session_id": s["id"],
        "run_id": run["id"],
        "agent_id": "writer",
        "tool_name": "create_ticket",
        "idempotency_key": f"{run['id']}:writer:1:tc-1",
        "risk_level": "R3",
        "status": "waiting_approval",
        "arguments": {"title": "test", "authorization": "***REDACTED***"},
    })
    await store.action_secrets.create(
        action_id=action["id"],
        arguments_encrypted=encrypt_params({"title": "test", "password": "real-secret-xyz"}),
    )

    # The action API response must NOT contain encrypted params.
    api_action = await svc.get_action(action_id=action["id"], user_id="u1")
    assert "arguments_encrypted" not in str(api_action).lower() or "arguments_encrypted" not in api_action
    assert "password" not in str(api_action.get("arguments", {}))

    # The encrypted params can only be decrypted server-side.
    decrypted = await coord.get_approved_action_params(action["id"])
    assert decrypted is not None
    assert decrypted["password"] == "real-secret-xyz"

    # But the decrypted params must never appear in the action's API response.
    api_action2 = await svc.get_action(action_id=action["id"], user_id="u1")
    assert "real-secret-xyz" not in json.dumps(api_action2)


# ── 16. ActionCoordinator: tool classification ───────────────────────────


@pytest.mark.asyncio
async def test_tool_classification_metadata_driven(store):
    """P1: tools are classified from capability snapshot metadata, not name
    prefixes. Tools without metadata use safe default (write/R3 →needs approval)."""
    snapshot = {
        "tools": [
            {"name": "search", "side_effect": "read", "risk_level": "R0"},
            {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
            {"name": "unknown_tool"},  # no metadata →safe default
        ],
    }

    # Read tool →no approval needed.
    assert classify_tool("search", snapshot) == ("read", "R0")
    assert not tool_needs_approval("search", "confirm_high_risk", snapshot)

    # Write tool R3 →approval needed in confirm_high_risk.
    assert classify_tool("create_ticket", snapshot) == ("write", "R3")
    assert tool_needs_approval("create_ticket", "confirm_high_risk", snapshot)

    # Unknown tool →safe default (write/R3) →approval needed.
    assert classify_tool("unknown_tool", snapshot) == ("write", "R3")
    assert tool_needs_approval("unknown_tool", "confirm_high_risk", snapshot)

    # In auto mode →no approval needed.
    assert not tool_needs_approval("create_ticket", "auto", snapshot)

    # In confirm_all_writes →any write needs approval.
    assert tool_needs_approval("create_ticket", "confirm_all_writes", snapshot)

    # No snapshot →safe default.
    assert classify_tool("any_tool", None) == ("write", "R3")


# ── 17. ActionCoordinator: encryption round-trip ─────────────────────────


def test_encryption_round_trip():
    """P1: encrypted params can be decrypted back correctly."""
    params = {"title": "test", "password": "secret123", "nested": {"key": "val"}}
    encrypted = encrypt_params(params)
    decrypted = decrypt_params(encrypted)
    assert decrypted == params
    # The ciphertext must not contain the plaintext.
    assert "secret123" not in encrypted
    assert "test" not in encrypted


# ── 18. waiting_approval does NOT write finished_at ──────────────────────


@pytest.mark.asyncio
async def test_waiting_approval_no_finished_at(store):
    """P1: waiting_approval is NOT a terminal state —finished_at must not
    be written, and both session and run should be in waiting_approval."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(
        runner.adapter,
        responses={"writer": "需要创建工单"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)

    final_run = await svc.get_run(run_id=run["id"], user_id="u1")
    assert final_run["status"] == "waiting_approval"
    assert final_run["finished_at"] is None

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


# ── 19. RoundtableState.from_dict is actually called by resume ────────────


@pytest.mark.asyncio
async def test_resume_uses_from_dict(store):
    """P1: RoundtableState.from_dict() is actually called by the resume path,
    not left idle."""
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "t",
            "write_policy": "confirm_all_writes",
            "participants": [{"agent_id": "writer", "role": "seat"}],
            "capability_snapshot": {
                "agents": [{"agent_id": "writer"}],
                "tools": [
                    {"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
                ],
            },
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")
    _patch_runtime_client(
        runner.adapter,
        responses={"writer": "工单已创建"},
        tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
    )

    await runner.execute(s, run)

    # Verify checkpoint was persisted.
    run_after = await svc.get_run(run_id=run["id"], user_id="u1")
    assert run_after["maf_checkpoint"] is not None
    assert run_after["maf_checkpoint"]["run_id"] == run["id"]

    # Approve and resume.
    approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
    await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
    resumed = await runner.resume(run["id"])
    assert resumed is not None
    assert resumed["status"] == "completed"


# ── 20. Token usage aggregation ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_token_usage_aggregated_on_completion(store):
    from app.maf_roundtable.workflow_runner import _aggregate_token_usage

    turns = [
        {"agent_id": "a", "turn_index": 1, "usage": {"input_tokens": 100, "output_tokens": 50, "total_tokens": 150}},
        {"agent_id": "b", "turn_index": 1, "usage": {"input_tokens": 200, "output_tokens": 80, "total_tokens": 280}},
        {"agent_id": "a", "turn_index": 2, "usage": None},
    ]
    agg = _aggregate_token_usage(turns)
    assert agg["total_input_tokens"] == 300
    assert agg["total_output_tokens"] == 130
    assert agg["total_tokens"] == 430
    assert len(agg["per_seat"]) == 2


# ── 21. Capability snapshot reading ──────────────────────────────────────


@pytest.mark.asyncio
async def test_capability_snapshot_drives_seat_config(store):
    from app.maf_roundtable.workflow_executors import _agent_capabilities_from_snapshot

    snapshot = {
        "agents": [
            {
                "agent_id": "seat-1",
                "model_name": "gpt-4o",
                "reasoning_effort": "high",
                "thinking_enabled": False,
                "subagent_enabled": True,
                "excluded_tools": ["dangerous_tool"],
                "memory_recall_disabled": True,
                "memory_injection_enabled": False,
            },
        ],
    }
    caps = _agent_capabilities_from_snapshot(snapshot, "seat-1")
    assert caps["model_name"] == "gpt-4o"
    assert caps["reasoning_effort"] == "high"
    assert caps["thinking_enabled"] is False
    assert caps["subagent_enabled"] is True
    assert caps["excluded_tools"] == ["dangerous_tool"]
    assert caps["memory_recall_disabled"] is True
    assert caps["memory_injection_enabled"] is False

    caps2 = _agent_capabilities_from_snapshot(snapshot, "unknown-agent")
    assert caps2 == {}

    caps3 = _agent_capabilities_from_snapshot(None, "seat-1")
    assert caps3 == {}


# ── 22. P1-3: Real WorkflowBuilder manager stages via WorkflowRunner ─────────


@pytest.mark.asyncio
async def test_real_workflow_manager_stages_persisted(store):
    """P1-3: drive a REAL MAF WorkflowRunner (real WorkflowBuilder, real SQLite
    repository, real adapter + seat-run store) through a structured run with a
    manager + 2 seats. Only the runtime client is faked (no real LLM).

    Asserts:
    - manager has planner, cross_review, finalizer seat-run rows
    - cross_review.turn_index == finalizer.turn_index
    - cross_review.execution_stage != finalizer.execution_stage
    """
    svc = _svc(store)
    s = await svc.create_session(
        user_id="u1",
        body={
            "task_text": "分析市场趋势",
            "mode": "structured",
            "manager_agent_id": "manager-1",
            "participants": [
                {"agent_id": "manager-1", "role": "manager"},
                {"agent_id": "seat-a", "role": "seat"},
                {"agent_id": "seat-b", "role": "seat"},
            ],
        },
    )
    run = await svc.create_run(session_id=s["id"], user_id="u1")

    app = _make_app(store)
    runner = WorkflowRunner(app, service=svc, user_id="u1")

    manager_responses = iter([
        "分工：seat-a 负责需求分析，seat-b 负责竞品分析",
        json.dumps({
            "summary": "审查通过，证据充分",
            "evidence_ok": True,
            "confirmed_candidates": [],
        }),
        "最终结论：市场看好，建议推进",
    ])
    seat_responses = {
        "seat-a": json.dumps({"summary": "需求旺盛", "findings": ["增长10%"]}),
        "seat-b": json.dumps({"summary": "竞品A领先", "findings": ["份额30%"]}),
    }

    call_log: list[dict] = []
    thread_counter = itertools.count()
    created_threads: list[str] = []
    latest_agent_by_thread: dict[str, str] = {}

    async def fake_create_thread(principal):
        thread_id = f"thread-{principal.session_id[:8]}-{next(thread_counter)}"
        created_threads.append(thread_id)
        return thread_id

    async def fake_start_agent_run(principal, *, thread_id, agent_id, message, **kwargs):
        latest_agent_by_thread[thread_id] = agent_id
        call_log.append({"agent_id": agent_id, "thread_id": thread_id, "message": message})
        return _FakeRunRecord(run_id=f"df-{agent_id}-{len(call_log)}")

    async def fake_read_messages(principal, thread_id):
        agent_id = latest_agent_by_thread[thread_id]
        if agent_id == "manager-1":
            content = next(manager_responses)
        else:
            content = seat_responses[agent_id]
        return [{"type": "ai", "content": content, "usage_metadata": {"input_tokens": 5, "output_tokens": 5}}]

    runner.adapter._client.create_thread = fake_create_thread
    runner.adapter._client.start_agent_run = fake_start_agent_run
    runner.adapter._client.read_thread_messages = fake_read_messages
    runner.adapter._client.cancel_run = AsyncMock(return_value=True)

    await runner.execute(s, run)

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

    seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
    mgr_rows = [r for r in seat_runs if r["agent_id"] == "manager-1"]
    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}"

    cr_rows = [r for r in mgr_rows if r["execution_stage"] == "cross_review"]
    fin_rows = [r for r in mgr_rows if r["execution_stage"] == "finalizer"]
    assert len(cr_rows) == 1 and len(fin_rows) == 1
    assert cr_rows[0]["turn_index"] == fin_rows[0]["turn_index"]
    assert cr_rows[0]["execution_stage"] != fin_rows[0]["execution_stage"]

    assert len(created_threads) == 3
    assert len(set(created_threads)) == 3

    ckpt = final_run["maf_checkpoint"]
    assert "最终结论：市场看好" in str(ckpt)
    assert any(
        r.get("agent_id") == "seat-a"
        and (r.get("structured_output") or {}).get("summary") == "需求旺盛"
        for r in ckpt["turn_results"]
    )