"""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"]) == []