226 lines
8.4 KiB
Python
226 lines
8.4 KiB
Python
"""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()
|