deerflow-code/offline-backend-20260512/backend/tests/test_maf_real_write_e2e.py

"""MAF approval-to-real-tool E2E coverage using the test outbox tool.

The tool is a real LangChain StructuredTool and is loaded through the normal
``get_available_tools`` + LangGraph ``ToolNode`` path.  It is intentionally
only added to this test's AppConfig; production ``config.yaml`` does not
register it.
"""

from __future__ import annotations

import json
from types import SimpleNamespace

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.tool_executor import ToolExecutor
from app.maf_roundtable.write_guard_bridge import (
    MafWriteGuardInterceptor,
    MafWriteGuardPolicy,
    register_write_guard_service,
)
from deerflow.agents.middlewares.write_guard_middleware import WriteGuardMiddleware
from deerflow.config import get_app_config
from deerflow.config.tool_config import ToolConfig
from deerflow.config.write_guard_config import WriteGuardConfig, WriteGuardProviderConfig
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import Base

TEST_TOOL = "maf_test_outbox_write"


@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)
    result = make_maf_roundtable_store(sf)
    assert result is not None
    yield result
    await engine.dispose()


@pytest.fixture
def svc(store):
    result = make_maf_service(store, enabled=True, admin_diagnostic_access=True)
    register_write_guard_service(result)
    return result


def _app_config():
    config = get_app_config().model_copy(deep=True)
    config.tools = [
        *config.tools,
        ToolConfig(
            name=TEST_TOOL,
            group="maf-e2e",
            use="deerflow.tools.builtins.maf_test_outbox:maf_test_outbox_write",
        ),
    ]
    config.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",
        ),
    )
    return config


def _snapshot():
    tool = {"name": TEST_TOOL, "side_effect": "write", "risk_level": "R2"}
    return {
        "tools": [tool],
        "agents": [{
            "agent_id": "seat-1",
            "allowed_tool_names": [TEST_TOOL],
            "excluded_tools": [],
            "tools": [tool],
        }],
    }


async def _make_pending_action(svc, store):
    session = await svc.create_session(user_id="u1", body={"task_text": "test"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    coordinator = ActionCoordinator(svc, user_id="u1")
    actions = await coordinator.intercept_write_tool_calls(
        run_id=run["id"],
        session_id=session["id"],
        agent_id="seat-1",
        turn_index=1,
        tool_calls=[{
            "name": TEST_TOOL,
            "id": "call-outbox-1",
            "args": {"operation_id": "op-1", "value": "controlled-value"},
        }],
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )
    assert len(actions) == 1
    return session, run, coordinator, actions[0]


@pytest.mark.asyncio
async def test_real_outbox_requires_approval_and_executes_once(monkeypatch, tmp_path, svc, store):
    """Pending action has no side effect; approved ToolNode call writes once."""
    monkeypatch.setenv("MAF_E2E_OUTBOX_DIR", str(tmp_path))
    monkeypatch.setattr(
        "deerflow.config.agents_config.load_agent_config",
        lambda _agent_id: SimpleNamespace(tool_groups=["maf-e2e"], model=None),
    )

    session, run, coordinator, action = await _make_pending_action(svc, store)
    outbox_file = tmp_path / "op-1.jsonl"
    assert not outbox_file.exists(), "approval interception must not execute the tool"

    assert await store.actions.mark_approved(action["id"]) is True

    class Adapter:
        class Client:
            async def execute_tool_directly(self, **kwargs):
                from deerflow.tools.approved_tool_executor import execute_approved_tool

                return await execute_approved_tool(
                    app_config=_app_config(),
                    checkpointer=None,
                    store=None,
                    thread_id=kwargs["thread_id"],
                    user_id="u1",
                    session_id=kwargs["principal"].session_id,
                    run_id=kwargs["principal"].run_id,
                    agent_id=kwargs["agent_id"],
                    turn_index=kwargs["turn_index"],
                    model_name=None,
                    excluded_tools=[],
                    capability_snapshot=kwargs["capability_snapshot"],
                    write_policy=kwargs["write_policy"],
                    tool_name=kwargs["tool_name"],
                    params=kwargs["params"],
                    approved_action_id=kwargs["approved_action_id"],
                    action_record=kwargs["action_record"],
                )

        def __init__(self):
            self._client = self.Client()

    executor = ToolExecutor(
        adapter=Adapter(),
        action_coordinator=coordinator,
        user_id="u1",
    )
    first = await executor.execute(
        action_id=action["id"],
        session_id=session["id"],
        run_id=run["id"],
        agent_id="seat-1",
        thread_id="thread-1",
        approved_tool=TEST_TOOL,
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )
    second = await executor.execute(
        action_id=action["id"],
        session_id=session["id"],
        run_id=run["id"],
        agent_id="seat-1",
        thread_id="thread-1",
        approved_tool=TEST_TOOL,
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )

    assert first["status"] == "succeeded"
    assert second["status"] == "succeeded"
    assert outbox_file.read_text(encoding="utf-8").count("\n") == 1
    stored = await store.actions.get_action(action["id"])
    assert stored["status"] == "succeeded"
    assert stored["result"]["tool_result"]["external_object_id"] == "op-1"


@pytest.mark.asyncio
async def test_rejected_action_never_reaches_real_tool(monkeypatch, tmp_path, svc, store):
    monkeypatch.setenv("MAF_E2E_OUTBOX_DIR", str(tmp_path))
    _session, _run, _coordinator, action = await _make_pending_action(svc, store)

    assert await store.actions.mark_rejected(action["id"]) is True
    assert not list(tmp_path.iterdir())
    assert (await store.actions.get_action(action["id"]))["status"] == "rejected"


@pytest.mark.asyncio
async def test_missing_secret_fails_closed_before_action_creation(monkeypatch, tmp_path, svc, store):
    monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False)
    monkeypatch.setenv("MAF_E2E_OUTBOX_DIR", str(tmp_path))
    session = await svc.create_session(user_id="u1", body={"task_text": "test"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    middleware = WriteGuardMiddleware(
        policy=MafWriteGuardPolicy(),
        interceptor=MafWriteGuardInterceptor(),
    )
    called = 0

    context = {
            "maf_run_id": run["id"],
            "maf_session_id": session["id"],
            "maf_user_id": "u1",
            "agent_id": "seat-1",
            "turn_index": 1,
            "write_policy": "confirm_all_writes",
            "capability_snapshot": _snapshot(),
        }
    request = SimpleNamespace(
        tool_call={
            "name": TEST_TOOL,
            "id": "call-missing-key",
            "args": {"operation_id": "op-missing-key", "value": "must-not-run"},
        },
        runtime=SimpleNamespace(context=context),
    )

    async def handler(_request):
        nonlocal called
        called += 1
        return {"executed": True}

    result = await middleware.awrap_tool_call(request, handler)

    envelope = json.loads(result.content)
    assert envelope["status"] == "intercept_failed"
    assert called == 0
    assert not list(tmp_path.iterdir())
    assert await store.actions.list_actions(run_id=run["id"]) == []