"""Acceptance tests for the WriteGuard middleware + MAF bridge (Task 1).

Verifies the core security invariants:
1. Write tools stay visible to the model (NOT excluded) but the real tool
   executes **0 times** before approval —the middleware intercepts and
   returns an ``approval_required`` ToolMessage instead.
2. After approval, the ToolExecutor runs the tool **exactly once** with the
   approved original params (model deviation is ignored).
3. Two concurrent ``resume()``/occupy calls →only one executor wins.
4. The single-transaction action bundle rolls back if any step fails.
5. ``auto`` policy →write tools pass through (no interception).
6. Legacy runs (no ``maf_run_id`` in context) →middleware is a no-op.

These tests exercise the middleware + interceptor in isolation (no LLM, no
DeerFlow runtime) so the security properties are deterministic and fast.
"""

from __future__ import annotations

import asyncio
import json
from collections.abc import Awaitable, Callable
from typing import Any
from unittest.mock import MagicMock

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.action_coordinator import ActionCoordinator
from app.maf_roundtable.write_guard_bridge import (
    MafWriteGuardInterceptor,
    MafWriteGuardPolicy,
    make_interceptor,
    make_policy,
    register_write_guard_service,
)
from deerflow.agents.middlewares.write_guard_middleware import WriteGuardMiddleware
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import Base


@pytest_asyncio.fixture
async def store():
    engine = create_async_engine("sqlite+aiosqlite://", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    s = make_maf_roundtable_store(sf)
    assert s is not None
    yield s
    await engine.dispose()


@pytest.fixture
def svc(store):

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


@pytest.fixture(autouse=True)
def _wire_bridge(svc):
    register_write_guard_service(svc)


def _capability_snapshot(*, write_policy: str = "confirm_all_writes") -> dict[str, Any]:
    """A snapshot with one read tool and one write tool."""
    return {
        "tools": [
            {"name": "read_data", "side_effect": "read", "risk_level": "R0"},
            {"name": "update_task", "side_effect": "write", "risk_level": "R3"},
        ],
    }


def _runtime_context(*, active: bool = True, write_policy: str = "confirm_all_writes") -> dict[str, Any]:
    """A run context mimicking what runtime_client.start_agent_run injects."""
    ctx: dict[str, Any] = {
        "agent_id": "seat-1",
        "turn_index": 1,
        "write_policy": write_policy,
        "capability_snapshot": _capability_snapshot(write_policy=write_policy),
    }
    if active:
        ctx["maf_run_id"] = "run-1"
        ctx["maf_session_id"] = "sess-1"
        ctx["maf_user_id"] = "u1"
    return ctx


def _tool_call_request(
    name: str,
    args: dict[str, Any] | None = None,
    *,
    run_context: dict[str, Any] | None = None,
):
    """Build a ToolCallRequest-like object with the fields the middleware reads."""
    runtime = MagicMock()
    runtime.context = run_context or {}
    req = MagicMock()
    req.tool_call = {"name": name, "args": args or {}, "id": f"call-{name}"}
    req.runtime = runtime
    return req


def _real_tool_handler(call_counter: dict[str, int]) -> Callable[[Any], Awaitable[Any]]:
    """An async handler that records each invocation —proves the real tool ran."""
    async def handler(request):
        name = request.tool_call["name"]
        call_counter[name] = call_counter.get(name, 0) + 1
        msg = MagicMock()
        msg.content = f"executed {name}"
        return msg
    return handler


# ── 1. Real tool executes 0 times before approval ──────────────────────────


@pytest.mark.asyncio
async def test_write_tool_never_executes_before_approval(svc, store):
    """The real tool handler is NOT called for a write tool awaiting approval.

    The middleware returns an approval_required ToolMessage and short-circuits.
    """
    call_counter: dict[str, int] = {}
    policy = MafWriteGuardPolicy()
    interceptor = MafWriteGuardInterceptor()
    mw = WriteGuardMiddleware(policy=policy, interceptor=interceptor)

    # Need a session + run for the bundle to reference.
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    await svc.create_run(session_id=session["id"], user_id="u1")
    ctx = _runtime_context()
    ctx["maf_session_id"] = session["id"]

    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    result = await mw.awrap_tool_call(request, handler)

    # Real tool never ran.
    assert call_counter.get("update_task", 0) == 0
    # An approval_required message was returned.
    envelope = json.loads(result.content)
    assert envelope["status"] == "approval_required"
    assert envelope["action_id"]
    assert envelope["tool_name"] == "update_task"


@pytest.mark.asyncio
async def test_read_tool_passes_through(svc, store):
    """A read tool is NOT intercepted —it executes normally even in an active run."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    request = _tool_call_request("read_data", {"q": "x"}, run_context=_runtime_context())
    await mw.awrap_tool_call(request, _real_tool_handler(call_counter))
    assert call_counter.get("read_data", 0) == 1


# ── 5. auto policy →no interception ────────────────────────────────────────


@pytest.mark.asyncio
async def test_auto_policy_write_tool_executes(svc, store):
    """Under ``auto`` policy, a write tool passes through without interception."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context(write_policy="auto")
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    await mw.awrap_tool_call(request, _real_tool_handler(call_counter))
    assert call_counter.get("update_task", 0) == 1


# ── 6. Legacy runs (no maf_run_id) →no-op ──────────────────────────────────


@pytest.mark.asyncio
async def test_legacy_run_noop():
    """Without ``maf_run_id`` in the context, the middleware passes through."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    request = _tool_call_request("update_task", {"title": "A"}, run_context=_runtime_context(active=False))
    await mw.awrap_tool_call(request, _real_tool_handler(call_counter))
    assert call_counter.get("update_task", 0) == 1


# ── 3. At-most-once: concurrent occupy →only one winner ────────────────────


@pytest.mark.asyncio
async def test_concurrent_occupy_only_one_winner(svc, store):
    """Two concurrent claim_for_execution calls →exactly one returns "claimed";
    the loser reflects the real state (in_progress), not a blanket skip (P0-3)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # Create an action in 'approved' state directly.
    action = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "k1",
        "risk_level": "R3", "status": "approved",
    })

    coord = ActionCoordinator(svc, user_id="u1")
    # Two concurrent occupies.
    r1, r2 = await asyncio.gather(
        coord.occupy_for_execution(action["id"]),
        coord.occupy_for_execution(action["id"]),
    )
    claimed = [r for r in (r1, r2) if r == "claimed"]
    assert len(claimed) == 1, f"expected exactly one 'claimed', got {(r1, r2)}"
    # The loser sees the action is already being executed (real state), so the
    # parent run can stay running instead of treating it as a no-op skip.
    losers = [r for r in (r1, r2) if r != "claimed"]
    assert losers == ["in_progress"], f"loser must reflect in_progress, got {losers}"


# ── 3b. record_terminal_result is state-machine constrained ─────────────────


@pytest.mark.asyncio
async def test_record_result_only_from_running(svc, store):
    """record_terminal_result only transitions running →terminal."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # Action in 'running' state.
    action = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "k2",
        "risk_level": "R3", "status": "running",
    })

    # First record →succeeds, returns the updated row.
    res1 = await store.actions.record_terminal_result(action["id"], status="succeeded", result={"ok": True})
    assert res1 is not None
    assert res1["status"] == "succeeded"

    # Second record (already terminal) →no-op, returns None, does NOT overwrite.
    res2 = await store.actions.record_terminal_result(action["id"], status="failed", result={"ok": False})
    assert res2 is None
    final = await store.actions.get_action(action["id"])
    assert final["status"] == "succeeded"  # not overwritten


# ── 2. ToolExecutor runs once with approved params (direct tool execution) ──


@pytest.mark.asyncio
async def test_tool_executor_runs_once_with_approved_params(svc, store):
    """The ToolExecutor calls execute_tool_directly exactly once with the
    approved params, and records the terminal result. The LLM is never involved.

    P0-2 fix: raw params never enter a prompt —they go directly to the tool
    function via execute_tool_directly(). The adapter.run_seat_turn() is NOT
    called (the LLM is bypassed).
    """
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # Seed an approved action with encrypted params {title: "A"}.
    from app.maf_roundtable.action_coordinator import encrypt_params
    coord = ActionCoordinator(svc, user_id="u1")
    enc = encrypt_params({"title": "A"})
    action = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "k3",
        "risk_level": "R3", "status": "approved",
    })
    await store.action_secrets.create(action_id=action["id"], arguments_encrypted=enc)

    # Mock client.execute_tool_directly —this is what ToolExecutor now calls.
    original_params_received = None
    exec_count = {"n": 0}
    async def _mock_execute_tool(*, thread_id, agent_id, tool_name, params, principal, **kwargs):
        nonlocal original_params_received
        original_params_received = params
        exec_count["n"] += 1
        return {"status": "succeeded", "result": {"output": f"called {tool_name}"}}

    mock_client = MagicMock()
    mock_client.execute_tool_directly = _mock_execute_tool
    mock_adapter = MagicMock()
    mock_adapter._client = mock_client

    from app.maf_roundtable.tool_executor import ToolExecutor
    executor = ToolExecutor(adapter=mock_adapter, action_coordinator=coord, user_id="u1")
    # P0-2: no approved_params kwarg —the executor recovers the params itself
    # from the encrypted secret stored above.
    outcome = await executor.execute(
        action_id=action["id"], session_id=session["id"], run_id=run["id"],
        agent_id="a1", thread_id="t1",
        approved_tool="update_task",
    )
    assert outcome["status"] == "succeeded"
    # The tool received the exact approved params (title=A, not something the
    # LLM modified).
    assert original_params_received == {"title": "A"}

    # The action is now terminal.
    final = await store.actions.get_action(action["id"])
    assert final["status"] == "succeeded"

    # A second execution attempt reflects the terminal state (P0-3: no blanket
    # "skipped") and —critically —does NOT re-execute the tool (at-most-once).
    outcome2 = await executor.execute(
        action_id=action["id"], session_id=session["id"], run_id=run["id"],
        agent_id="a1", thread_id="t1",
        approved_tool="update_task",
    )
    assert outcome2["status"] == "succeeded"  # reflects the real terminal status
    assert outcome2.get("action") is not None
    assert exec_count["n"] == 1, "tool must execute exactly once (at-most-once)"


# ── 2b. P0-2: secret unavailable ⇀hard fail, zero tool executions ─────────


@pytest.mark.asyncio
async def test_executor_missing_secret_fails_without_calling_tool(svc, store):
    """P0-2: an approved action whose secret row is missing must FAIL with the
    tool executing **0 times** —never fall back to empty params."""
    from app.maf_roundtable.action_coordinator import encrypt_params
    from app.maf_roundtable.tool_executor import ToolExecutor

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    coord = ActionCoordinator(svc, user_id="u1")

    action = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "k-missing",
        "risk_level": "R3", "status": "approved",
    })
    # Seed then DELETE the secret so the row is gone.
    await store.action_secrets.create(
        action_id=action["id"], arguments_encrypted=encrypt_params({"title": "A"})
    )
    await store.action_secrets.delete_by_action(action["id"])
    assert await store.action_secrets.get_by_action(action["id"]) is None

    calls = {"n": 0}

    async def _boom(*, thread_id, agent_id, tool_name, params, principal):
        calls["n"] += 1
        return {"status": "succeeded", "result": {}}

    mock_client = MagicMock()
    mock_client.execute_tool_directly = _boom
    mock_adapter = MagicMock()
    mock_adapter._client = mock_client

    executor = ToolExecutor(adapter=mock_adapter, action_coordinator=coord, user_id="u1")
    outcome = await executor.execute(
        action_id=action["id"], session_id=session["id"], run_id=run["id"],
        agent_id="a1", thread_id="t1", approved_tool="update_task",
    )

    assert calls["n"] == 0, "tool must NOT run when the secret is missing"
    assert outcome["status"] == "failed"
    final = await store.actions.get_action(action["id"])
    assert final["status"] == "failed"
    assert final["status"] != "succeeded"


@pytest.mark.asyncio
async def test_executor_corrupt_secret_fails_without_calling_tool(svc, store):
    """P0-2: an undecryptable secret (wrong key / garbage) must FAIL with the
    tool executing **0 times** —not run with empty params."""
    from app.maf_roundtable.tool_executor import ToolExecutor

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    coord = ActionCoordinator(svc, user_id="u1")

    action = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "k-corrupt",
        "risk_level": "R3", "status": "approved",
    })
    # Garbage ciphertext that Fernet cannot decrypt.
    await store.action_secrets.create(
        action_id=action["id"], arguments_encrypted="not-a-valid-fernet-token"
    )

    calls = {"n": 0}

    async def _boom(*, thread_id, agent_id, tool_name, params, principal):
        calls["n"] += 1
        return {"status": "succeeded", "result": {}}

    mock_client = MagicMock()
    mock_client.execute_tool_directly = _boom
    mock_adapter = MagicMock()
    mock_adapter._client = mock_client

    executor = ToolExecutor(adapter=mock_adapter, action_coordinator=coord, user_id="u1")
    outcome = await executor.execute(
        action_id=action["id"], session_id=session["id"], run_id=run["id"],
        agent_id="a1", thread_id="t1", approved_tool="update_task",
    )

    assert calls["n"] == 0, "tool must NOT run when the secret cannot be decrypted"
    assert outcome["status"] == "failed"
    final = await store.actions.get_action(action["id"])
    assert final["status"] == "failed"


@pytest.mark.asyncio
async def test_get_approved_action_params_raises_with_reasons(svc, store):
    """P0-2: get_approved_action_params raises ActionSecretUnavailableError
    (never returns None) with a distinct reason per failure mode."""
    from app.maf_roundtable.action_coordinator import (
        ActionSecretUnavailableError,
        encrypt_params,
    )

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    coord = ActionCoordinator(svc, user_id="u1")

    # missing
    a_missing = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "r-missing",
        "risk_level": "R3", "status": "approved",
    })
    with pytest.raises(ActionSecretUnavailableError) as ei:
        await coord.get_approved_action_params(a_missing["id"])
    assert ei.value.reason == "missing"

    # decrypt_failed
    a_corrupt = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "r-corrupt",
        "risk_level": "R3", "status": "approved",
    })
    await store.action_secrets.create(action_id=a_corrupt["id"], arguments_encrypted="garbage")
    with pytest.raises(ActionSecretUnavailableError) as ei:
        await coord.get_approved_action_params(a_corrupt["id"])
    assert ei.value.reason == "decrypt_failed"

    # invalid_type —valid Fernet token whose plaintext is a JSON list, not object
    from cryptography.fernet import Fernet

    from app.maf_roundtable.action_coordinator import _get_fernet_key

    list_token = Fernet(_get_fernet_key()).encrypt(b"[1, 2, 3]").decode()
    a_list = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "r-list",
        "risk_level": "R3", "status": "approved",
    })
    await store.action_secrets.create(action_id=a_list["id"], arguments_encrypted=list_token)
    with pytest.raises(ActionSecretUnavailableError) as ei:
        await coord.get_approved_action_params(a_list["id"])
    assert ei.value.reason == "invalid_type"

    # happy path still returns the dict
    a_ok = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "a1",
        "tool_name": "update_task", "idempotency_key": "r-ok",
        "risk_level": "R3", "status": "approved",
    })
    await store.action_secrets.create(
        action_id=a_ok["id"], arguments_encrypted=encrypt_params({"title": "A"})
    )
    assert await coord.get_approved_action_params(a_ok["id"]) == {"title": "A"}


# ── 4. Single-transaction bundle atomicity + idempotency ──────────────────


@pytest.mark.asyncio
async def test_action_bundle_writes_all_four_atomically(svc, store):
    """A successful bundle writes action + secret + approval + audit event
    together —none can exist without the others (single transaction)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    coord = ActionCoordinator(svc, user_id="u1")
    created = await coord.intercept_write_tool_calls(
        run_id=run["id"], session_id=session["id"], agent_id="a1", turn_index=1,
        tool_calls=[{"name": "update_task", "id": "c1", "args": {"title": "A"}}],
        capability_snapshot=_capability_snapshot(), write_policy="confirm_all_writes",
    )
    assert len(created) == 1
    action_id = created[0]["id"]

    # All four rows exist (atomic success).
    action = await store.actions.get_action(action_id)
    assert action is not None and action["status"] == "waiting_approval"
    assert action["approval_id"], "approval must be linked"

    secret = await store.action_secrets.get_by_action(action_id)
    assert secret is not None, "encrypted secret must exist"

    approval = await store.approvals.get_approval(action["approval_id"])
    assert approval is not None and approval["status"] == "pending"

    events = await store.events.list_events(run_id=run["id"])
    assert any(e["event_type"] == "action.waiting_approval" for e in events), \
        "audit event must exist"


@pytest.mark.asyncio
async def test_action_bundle_idempotent_on_duplicate(svc, store):
    """A duplicate (same idempotency_key) submission re-reads and returns the
    existing action instead of creating a second one —retries never fail the run."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    coord = ActionCoordinator(svc, user_id="u1")
    args = [dict(run_id=run["id"], session_id=session["id"], agent_id="a1", turn_index=1,
                 tool_calls=[{"name": "update_task", "id": "c1", "args": {"title": "A"}}],
                 capability_snapshot=_capability_snapshot(), write_policy="confirm_all_writes")]

    first = await coord.intercept_write_tool_calls(**args[0])
    second = await coord.intercept_write_tool_calls(**args[0])

    assert first[0]["id"] == second[0]["id"], "duplicate must return the same action id"
    actions = await store.actions.list_actions(run_id=run["id"])
    assert len(actions) == 1, f"expected 1 action, got {len(actions)}"


# ── Factory resolution ──────────────────────────────────────────────────────


def test_factories_return_singletons():
    """make_policy/make_interceptor return the bridge objects."""
    p = make_policy()
    i = make_interceptor()
    assert isinstance(p, MafWriteGuardPolicy)
    assert isinstance(i, MafWriteGuardInterceptor)


# ── P0-3: Strengthened WriteGuard passthrough ───────────────────────────────


@pytest.mark.asyncio
async def test_approved_passthrough_requires_action_id(svc, store):
    """When maf_approved_tool is set but maf_approved_action_id is missing,
    the WriteGuard blocks the passthrough —tool name matching alone is
    insufficient (P0-3 fix)."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()
    ctx["maf_approved_tool"] = "update_task"
    # Deliberately omit maf_approved_action_id —should be blocked.
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    result = await mw.awrap_tool_call(request, handler)

    # Real tool should NOT execute —the passthrough requires action_id.
    assert call_counter.get("update_task", 0) == 0
    # Should return an approval_required or error message, not pass through.
    envelope = json.loads(result.content)
    assert envelope["status"] in ("approval_required", "intercept_failed")


@pytest.mark.asyncio
async def test_approved_passthrough_allows_with_action_id(svc, store):
    """When both maf_approved_tool and maf_approved_action_id are set AND the
    bound action validates (status running + run/session/agent + tool match),
    the WriteGuard allows the passthrough (defense-in-depth)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    await svc.create_run(session_id=session["id"], user_id="u1")
    # Seed the bound action in the state the executor leaves it (running) with
    # ownership matching the run context —validate_approved_action queries it.
    await store.actions.create_action({
        "id": "action-1", "session_id": session["id"], "run_id": "run-1",
        "agent_id": "seat-1", "tool_name": "update_task",
        "idempotency_key": "pt-allow", "risk_level": "R3", "status": "running",
    })
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()
    ctx["maf_session_id"] = session["id"]
    ctx["maf_approved_tool"] = "update_task"
    ctx["maf_approved_action_id"] = "action-1"
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    await mw.awrap_tool_call(request, handler)

    # Real tool should execute —passthrough is active and validated.
    assert call_counter.get("update_task", 0) == 1


@pytest.mark.asyncio
async def test_approved_passthrough_blocks_when_action_not_running(svc, store):
    """P0-4: field presence is not enough —if the bound action is NOT in
    ``running`` status, the async passthrough fails closed (tool does not run)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    await svc.create_run(session_id=session["id"], user_id="u1")
    # Action exists but is still waiting_approval (not occupied/running).
    await store.actions.create_action({
        "id": "action-2", "session_id": session["id"], "run_id": "run-1",
        "agent_id": "seat-1", "tool_name": "update_task",
        "idempotency_key": "pt-notrunning", "risk_level": "R3", "status": "waiting_approval",
    })
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()
    ctx["maf_session_id"] = session["id"]
    ctx["maf_approved_tool"] = "update_task"
    ctx["maf_approved_action_id"] = "action-2"
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    result = await mw.awrap_tool_call(request, handler)

    assert call_counter.get("update_task", 0) == 0, "must fail closed when action is not running"
    envelope = json.loads(result.content)
    assert envelope["status"] == "approval_invalid"


@pytest.mark.asyncio
async def test_approved_passthrough_blocks_when_action_ownership_mismatches(svc, store):
    """P0-4: a running action belonging to a DIFFERENT run must not validate."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    await svc.create_run(session_id=session["id"], user_id="u1")
    await store.actions.create_action({
        "id": "action-3", "session_id": session["id"], "run_id": "OTHER-RUN",
        "agent_id": "seat-1", "tool_name": "update_task",
        "idempotency_key": "pt-ownership", "risk_level": "R3", "status": "running",
    })
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()  # maf_run_id = "run-1"
    ctx["maf_session_id"] = session["id"]
    ctx["maf_approved_tool"] = "update_task"
    ctx["maf_approved_action_id"] = "action-3"
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    result = await mw.awrap_tool_call(request, handler)

    assert call_counter.get("update_task", 0) == 0
    envelope = json.loads(result.content)
    assert envelope["status"] == "approval_invalid"


@pytest.mark.asyncio
async def test_approved_passthrough_blocks_different_tool(svc, store):
    """When maf_approved_tool is 'update_task' but the LLM calls 'delete_task',
    the passthrough blocks —different tool name (defense-in-depth)."""
    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    await svc.create_run(session_id=session["id"], user_id="u1")
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()
    ctx["maf_session_id"] = session["id"]
    ctx["maf_approved_tool"] = "update_task"
    ctx["maf_approved_action_id"] = "action-1"
    request = _tool_call_request("delete_task", {"id": "x"}, run_context=ctx)
    handler = _real_tool_handler(call_counter)

    result = await mw.awrap_tool_call(request, handler)

    assert call_counter.get("delete_task", 0) == 0
    envelope = json.loads(result.content)
    assert envelope["status"] == "approval_required"


# ── P0-4: sync path distinguishes read (pass) vs write (fail-closed) ────────


def _sync_tool_handler(call_counter: dict[str, int]):
    """A sync handler that records each invocation —proves the real tool ran."""
    def handler(request):
        name = request.tool_call["name"]
        call_counter[name] = call_counter.get(name, 0) + 1
        msg = MagicMock()
        msg.content = f"executed {name}"
        return msg
    return handler


def test_sync_read_tool_passes_through(svc, store):
    """P0-4: on the SYNC path a read tool is never intercepted —it runs."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()  # confirm_all_writes; read_data is side_effect=read
    request = _tool_call_request("read_data", {"q": "x"}, run_context=ctx)
    handler = _sync_tool_handler(call_counter)

    mw.wrap_tool_call(request, handler)

    assert call_counter.get("read_data", 0) == 1, "read tools must pass the sync path"


def test_sync_write_tool_fails_closed(svc, store):
    """P0-4: on the SYNC path a write tool that cannot be async-intercepted
    fails closed (blocked), it does NOT run unapproved."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()  # confirm_all_writes; update_task is side_effect=write
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _sync_tool_handler(call_counter)

    result = mw.wrap_tool_call(request, handler)

    assert call_counter.get("update_task", 0) == 0, "write tools must fail closed on the sync path"
    envelope = json.loads(result.content)
    assert envelope["status"] == "intercept_blocked_sync"


def test_sync_approved_passthrough_runs(svc, store):
    """P0-4: an approved deterministic execution (maf_approved_tool bound)
    passes the sync path without being blocked as a write."""
    call_counter: dict[str, int] = {}
    mw = WriteGuardMiddleware(policy=MafWriteGuardPolicy(), interceptor=MafWriteGuardInterceptor())
    ctx = _runtime_context()
    ctx["maf_approved_tool"] = "update_task"
    ctx["maf_approved_action_id"] = "action-sync"
    request = _tool_call_request("update_task", {"title": "A"}, run_context=ctx)
    handler = _sync_tool_handler(call_counter)

    mw.wrap_tool_call(request, handler)

    assert call_counter.get("update_task", 0) == 1, "approved passthrough must run on the sync path"


# ── P0-4: startup prerequisite validation ───────────────────────────────────


def test_validate_maf_prerequisites_requires_write_guard(monkeypatch):
    """validate_maf_prerequisites raises when write_guard is disabled."""
    from app.maf_roundtable.service import (
        MafConfigurationError,
        validate_maf_prerequisites,
    )
    from deerflow.config.write_guard_config import (
        load_write_guard_config_from_dict,
        reset_write_guard_config,
    )

    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "x" * 32)
    try:
        load_write_guard_config_from_dict({"enabled": False})
        with pytest.raises(MafConfigurationError):
            validate_maf_prerequisites()
    finally:
        reset_write_guard_config()


def test_validate_maf_prerequisites_passes_when_all_met(monkeypatch):
    """validate_maf_prerequisites passes when write_guard + provider + key present."""
    from app.maf_roundtable.service import validate_maf_prerequisites
    from deerflow.config.write_guard_config import (
        load_write_guard_config_from_dict,
        reset_write_guard_config,
    )

    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "x" * 32)
    try:
        load_write_guard_config_from_dict({
            "enabled": True,
            "provider": {
                "policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
                "interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
            },
        })
        validate_maf_prerequisites()  # must not raise
    finally:
        reset_write_guard_config()


def test_validate_maf_prerequisites_allows_read_only_without_secret(monkeypatch):
    """A missing key leaves reads available; writes fail closed at interception."""
    from app.maf_roundtable.service import validate_maf_prerequisites
    from deerflow.config.write_guard_config import (
        load_write_guard_config_from_dict,
        reset_write_guard_config,
    )

    monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False)
    try:
        load_write_guard_config_from_dict({
            "enabled": True,
            "provider": {
                "policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
                "interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
            },
        })
        validate_maf_prerequisites()
    finally:
        reset_write_guard_config()


def test_write_guard_health_reports_missing_secret(monkeypatch):
    from app.maf_roundtable.service import get_maf_write_guard_readiness
    from deerflow.config.write_guard_config import (
        load_write_guard_config_from_dict,
        reset_write_guard_config,
    )

    monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False)
    try:
        load_write_guard_config_from_dict({
            "enabled": True,
            "provider": {
                "policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
                "interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
            },
        })
        readiness = get_maf_write_guard_readiness()
        assert readiness == {
            "enabled": True,
            "ready": False,
            "secret_configured": False,
            "status": "missing_secret",
        }
    finally:
        reset_write_guard_config()


# ── P1b: SeatExecutor precise pending-approval matching ─────────────────────


class _FakeSeatResponse:
    def __init__(self, tool_calls, content="ok", error=None, usage=None):
        self.tool_calls = tool_calls
        self.content = content
        self.error = error
        self.usage = usage or {}


class _FakeSeatAdapter:
    """Minimal adapter: returns a canned seat response, exposes a real store."""

    def __init__(self, store, response):
        from types import SimpleNamespace

        self._service = SimpleNamespace(store=store)
        self._response = response

    async def run_seat_turn(self, **kwargs):
        return self._response

    async def _emit_critical(self, *args, **kwargs):
        pass


class _FakeSeatCtx:
    def __init__(self):
        self.sent = []

    async def send_message(self, msg):
        self.sent.append(msg)


def _seat_state(session_id, run_id):
    from app.maf_roundtable.workflow_executors import RoundtableState

    state = RoundtableState(
        session_id=session_id, run_id=run_id, task_text="t", mode="structured",
        model_name=None, reasoning_effort=None, max_rounds=8,
        write_policy="confirm_all_writes", manager_agent_id=None,
        seat_agent_ids=["seat-1"], thread_map={"seat-1": "thread-x"},
        capability_snapshot=_capability_snapshot(),
    )
    state.current_round = 1
    return state


@pytest.mark.asyncio
async def test_seat_matching_scopes_by_turn_index(svc, store):
    """P1b: a stale waiting action from an EARLIER turn is NOT mixed into the
    current turn's pending approval —matching keys on turn_index + tool_call_id."""
    from app.maf_roundtable.workflow_executors import SeatExecutor

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # Stale action from turn 0 (same agent/run, different turn + call id).
    stale = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "seat-1",
        "tool_name": "update_task", "turn_index": 0, "logical_action_id": "old-call",
        "idempotency_key": "stale", "risk_level": "R3", "status": "waiting_approval",
    })
    # Current action from turn 1 matching the response's tool_call id.
    current = await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "seat-1",
        "tool_name": "update_task", "turn_index": 1, "logical_action_id": "call-1",
        "idempotency_key": "current", "risk_level": "R3", "status": "waiting_approval",
    })

    state = _seat_state(session["id"], run["id"])
    response = _FakeSeatResponse(tool_calls=[{"name": "update_task", "id": "call-1", "args": {}}])
    executor = SeatExecutor(_FakeSeatAdapter(store, response), state, "seat-1",
                            action_coordinator=ActionCoordinator(svc, user_id="u1"))
    await executor.handle("msg", _FakeSeatCtx())

    assert len(state.pending_approvals) == 1
    assert state.pending_approvals[0]["action_id"] == current["id"]
    assert state.pending_approvals[0]["action_id"] != stale["id"]
    assert state.pending_approvals[0]["turn_index"] == 1


@pytest.mark.asyncio
async def test_seat_matching_zero_hits_warns_and_skips(svc, store):
    """A pending tool call that matches 0 waiting actions (e.g. unclassified
    tool, hallucinated tool name, or WriteGuard let it through as read) should
    log a warning and skip — NOT crash the entire workflow."""
    from app.maf_roundtable.workflow_executors import SeatExecutor

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # One waiting action for turn 1, but its logical_action_id != the response's.
    await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "seat-1",
        "tool_name": "update_task", "turn_index": 1, "logical_action_id": "other-call",
        "idempotency_key": "mismatch", "risk_level": "R3", "status": "waiting_approval",
    })

    state = _seat_state(session["id"], run["id"])
    response = _FakeSeatResponse(tool_calls=[{"name": "update_task", "id": "call-1", "args": {}}])
    executor = SeatExecutor(_FakeSeatAdapter(store, response), state, "seat-1",
                            action_coordinator=ActionCoordinator(svc, user_id="u1"))

    # Should NOT raise — the unmatched tool call is logged and skipped.
    await executor.handle("msg", _FakeSeatCtx())
    # The unmatched call was not added to pending_approvals.
    assert len(state.pending_approvals) == 0


@pytest.mark.asyncio
async def test_seat_matching_multiple_hits_raises(svc, store):
    """>1 match for a single tool call is a genuine data-integrity error —
    the executor refuses to guess and raises RuntimeError."""
    from app.maf_roundtable.workflow_executors import SeatExecutor

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    # Two waiting actions with the same logical_action_id for turn 1.
    await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "seat-1",
        "tool_name": "update_task", "turn_index": 1, "logical_action_id": "call-1",
        "idempotency_key": "dup1", "risk_level": "R3", "status": "waiting_approval",
    })
    await store.actions.create_action({
        "session_id": session["id"], "run_id": run["id"], "agent_id": "seat-1",
        "tool_name": "update_task", "turn_index": 1, "logical_action_id": "call-1",
        "idempotency_key": "dup2", "risk_level": "R3", "status": "waiting_approval",
    })

    state = _seat_state(session["id"], run["id"])
    response = _FakeSeatResponse(tool_calls=[{"name": "update_task", "id": "call-1", "args": {}}])
    executor = SeatExecutor(_FakeSeatAdapter(store, response), state, "seat-1",
                            action_coordinator=ActionCoordinator(svc, user_id="u1"))

    with pytest.raises(RuntimeError, match="refusing to guess"):
        await executor.handle("msg", _FakeSeatCtx())


# ── P0-1: WriteGuard middleware loads from config ────────────────────────────


def test_write_guard_middleware_in_chain_when_enabled():
    """When write_guard.enabled is True and provider is configured,
    build_lead_runtime_middlewares includes WriteGuardMiddleware."""
    from deerflow.agents.middlewares.tool_error_handling_middleware import (
        _build_runtime_middlewares,
    )
    from deerflow.config.write_guard_config import (
        WriteGuardConfig,
        WriteGuardProviderConfig,
        load_write_guard_config_from_dict,
        reset_write_guard_config,
    )

    try:
        load_write_guard_config_from_dict({
            "enabled": True,
            "provider": {
                "policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
                "interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
            },
        })
        # Build a minimal AppConfig with write_guard enabled.
        # Use model_construct to skip full validation (we only need
        # write_guard and sandbox fields for this middleware test).
        from deerflow.config import get_app_config
        cfg = get_app_config().__class__.model_construct(
            sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"},
            write_guard=WriteGuardConfig(
                enabled=True,
                provider=WriteGuardProviderConfig(
                    policy_use="app.maf_roundtable.write_guard_bridge:make_policy",
                    interceptor_use="app.maf_roundtable.write_guard_bridge:make_interceptor",
                ),
            ),
        )
        middlewares = _build_runtime_middlewares(
            app_config=cfg,
            include_uploads=False,
            include_dangling_tool_call_patch=False,
        )
        from deerflow.agents.middlewares.write_guard_middleware import WriteGuardMiddleware
        assert any(isinstance(mw, WriteGuardMiddleware) for mw in middlewares), (
            "WriteGuardMiddleware should be in the chain when write_guard is enabled"
        )
    finally:
        reset_write_guard_config()