"""Real HTTP approval → dispatcher → runner → ToolNode/outbox acceptance test.""" from __future__ import annotations from types import SimpleNamespace import pytest from fastapi import FastAPI, Request from httpx import ASGITransport, AsyncClient from maf_test_helpers import make_maf_service from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.gateway.routers.maf_roundtable import router as maf_router from app.maf_roundtable.action_coordinator import ActionCoordinator from app.maf_roundtable.dispatcher import get_dispatcher from app.maf_roundtable.runtime_client import DeerFlowRuntimeClient from app.maf_roundtable.workflow_executors import RoundtableState from app.maf_roundtable.write_guard_bridge import register_write_guard_service from deerflow.config import get_app_config from deerflow.config.maf_roundtable_config import MafRoundtableConfig from deerflow.config.tool_config import ToolConfig from deerflow.config.write_guard_config import WriteGuardConfig, WriteGuardProviderConfig from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.tools.approved_tool_executor import execute_approved_tool TEST_TOOL = "maf_test_outbox_write" class _User: def __init__(self, user_id: str) -> None: self.id = user_id self.system_role = "user" def _snapshot() -> dict: tool = { "name": TEST_TOOL, "side_effect": "write", "risk_level": "R2", "target_key_fields": ["operation_id"], "target_key_namespace": "maf-e2e", } return { "tools": [tool], "agents": [{ "agent_id": "seat-1", "allowed_tool_names": [TEST_TOOL], "excluded_tools": [], "tools": [tool], }], } 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 @pytest.mark.asyncio async def test_http_approval_dispatches_real_tool_once(monkeypatch, tmp_path): """A real route decision reaches the ToolNode exactly once, even on replay.""" monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "test-only-http-e2e-secret") 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), ) engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_http_e2e.db'}") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) sf = async_sessionmaker(engine, expire_on_commit=False) store = make_maf_roundtable_store(sf) assert store is not None svc = make_maf_service( store, enabled=True, config=MafRoundtableConfig(enabled=True, approval_timeout_seconds=300), ) register_write_guard_service(svc) app = FastAPI() app.state.maf_roundtable_service = svc app.include_router(maf_router) @app.middleware("http") async def test_auth(request: Request, call_next): request.state.user = _User("u1") return await call_next(request) async def execute_directly(_client, **kwargs): return await execute_approved_tool( app_config=_app_config(), checkpointer=None, store=None, thread_id=kwargs["thread_id"], user_id=kwargs["principal"].user_id, 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"], ) async def no_model_finalizer(_adapter, state): state.final_result = {"content": "http e2e final report"} monkeypatch.setattr(DeerFlowRuntimeClient, "execute_tool_directly", execute_directly) monkeypatch.setattr("app.maf_roundtable.workflow_runner.generate_final_report", no_model_finalizer) session = await svc.create_session(user_id="u1", body={"task_text": "HTTP E2E"}) 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": "http-outbox-call", "args": {"operation_id": "http-op-1", "value": "controlled"}, }], capability_snapshot=_snapshot(), write_policy="confirm_all_writes", ) action = actions[0] state = RoundtableState( session_id=session["id"], run_id=run["id"], task_text="HTTP E2E", mode="structured", model_name=None, reasoning_effort=None, max_rounds=1, write_policy="confirm_all_writes", manager_agent_id=None, seat_agent_ids=["seat-1"], thread_map={"seat-1": "thread-1"}, capability_snapshot=_snapshot(), ) state.current_round = 1 state.status = "waiting_approval" state.pending_approvals = [{ "action_id": action["id"], "approval_id": action["approval_id"], "agent_id": "seat-1", "tool_name": TEST_TOOL, }] await store.runs.update_run(run["id"], {"status": "waiting_approval", "maf_checkpoint": state.to_dict()}) await store.sessions.update_session( session["id"], user_id="u1", fields={"status": "waiting_approval"} ) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: listed = await client.get(f"/api/maf-roundtable/runs/{run['id']}/actions") assert listed.status_code == 200 listed_action = listed.json()["actions"][0] assert "target_key" not in listed_action assert "idempotency_key" not in listed_action assert "internal_canary" not in listed_action fetched = await client.get(f"/api/maf-roundtable/actions/{action['id']}") assert fetched.status_code == 200 fetched_action = fetched.json() assert "target_key" not in fetched_action assert "idempotency_key" not in fetched_action assert "internal_canary" not in fetched_action # The lease arbiter still needs the private key in persistence. stored_before_decision = await store.actions.get_action(action["id"]) assert stored_before_decision["target_key"] assert stored_before_decision["idempotency_key"] response = await client.post( f"/api/maf-roundtable/approvals/{action['approval_id']}/decision", json={"decision": "approved"}, ) assert response.status_code == 200 assert response.json()["status"] == "approved" duplicate = await client.post( f"/api/maf-roundtable/approvals/{action['approval_id']}/decision", json={"decision": "approved"}, ) assert duplicate.status_code == 409 outbox = tmp_path / "http-op-1.jsonl" assert outbox.read_text(encoding="utf-8").count("\n") == 1 stored = await store.actions.get_action(action["id"]) assert stored["status"] == "succeeded" assert stored["attempt_count"] == 1 assert (await store.runs.get_run(run["id"]))["status"] == "completed" # A process/retry replay after the terminal transition returns the stored # run and never reaches ToolNode again. replay = await get_dispatcher(app).resume(run["id"], user_id="u1") assert replay is not None and replay["status"] == "completed" assert outbox.read_text(encoding="utf-8").count("\n") == 1 await engine.dispose()