350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""Entry-level MAF Phase-5 approval and target-lease acceptance tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, Request
|
|
from httpx import ASGITransport, AsyncClient
|
|
from langchain_core.language_models.chat_models import BaseChatModel
|
|
from langchain_core.messages import AIMessage
|
|
from langchain_core.outputs import ChatGeneration, ChatResult
|
|
from langgraph.checkpoint.memory import InMemorySaver
|
|
from sqlalchemy import update
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.app import _scan_maf_approval_expiry
|
|
from app.gateway.routers.maf_roundtable import router as maf_router
|
|
from app.maf_roundtable.dispatcher import get_dispatcher
|
|
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, MafToolMetadata
|
|
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.persistence.maf_roundtable.model import MafRoundtableApprovalRow
|
|
from deerflow.runtime import RunManager
|
|
from deerflow.runtime.events.store.memory import MemoryRunEventStore
|
|
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
|
|
|
|
TEST_TOOL = "maf_test_outbox_write"
|
|
|
|
|
|
class _User:
|
|
def __init__(self, user_id: str) -> None:
|
|
self.id = user_id
|
|
self.system_role = "user"
|
|
|
|
|
|
class _ThreadStore:
|
|
def __init__(self) -> None:
|
|
self.rows: dict[str, dict] = {}
|
|
|
|
async def create(self, thread_id, assistant_id=None, metadata=None):
|
|
self.rows[thread_id] = {
|
|
"thread_id": thread_id,
|
|
"assistant_id": assistant_id,
|
|
"metadata": metadata or {},
|
|
"status": "idle",
|
|
}
|
|
|
|
async def get(self, thread_id):
|
|
return self.rows.get(thread_id)
|
|
|
|
async def update_status(self, thread_id, status):
|
|
self.rows.setdefault(thread_id, {})["status"] = status
|
|
|
|
async def update_assistant_id(self, thread_id, assistant_id):
|
|
self.rows.setdefault(thread_id, {})["assistant_id"] = assistant_id
|
|
|
|
async def update_metadata(self, thread_id, metadata):
|
|
self.rows.setdefault(thread_id, {}).setdefault("metadata", {}).update(metadata)
|
|
|
|
async def update_display_name(self, thread_id, display_name):
|
|
self.rows.setdefault(thread_id, {})["display_name"] = display_name
|
|
|
|
|
|
class _DeterministicModel(BaseChatModel):
|
|
"""Deterministic model boundary: propose one write, then finish."""
|
|
|
|
calls: int = 0
|
|
|
|
@property
|
|
def _llm_type(self) -> str:
|
|
return "maf-phase5-deterministic"
|
|
|
|
def bind_tools(self, tools, **kwargs): # noqa: ARG002
|
|
return self
|
|
|
|
def _generate(self, messages, stop=None, run_manager=None, **kwargs): # noqa: ARG002
|
|
if self.calls == 0:
|
|
message = AIMessage(
|
|
content="",
|
|
tool_calls=[{
|
|
"name": TEST_TOOL,
|
|
"args": {"operation_id": "phase5-target", "value": "controlled"},
|
|
"id": "phase5-tool-call",
|
|
"type": "tool_call",
|
|
}],
|
|
)
|
|
else:
|
|
message = AIMessage(content="deterministic completion")
|
|
self.calls += 1
|
|
return ChatResult(generations=[ChatGeneration(message=message)])
|
|
|
|
|
|
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",
|
|
),
|
|
)
|
|
config.maf_roundtable = MafRoundtableConfig(
|
|
enabled=True,
|
|
default_write_policy="confirm_all_writes",
|
|
approval_timeout_seconds=300,
|
|
approval_scan_interval_seconds=5,
|
|
tool_metadata={
|
|
TEST_TOOL: MafToolMetadata(
|
|
side_effect="write",
|
|
risk_level="R2",
|
|
target_key_fields=["operation_id"],
|
|
target_key_namespace="phase5-e2e",
|
|
)
|
|
},
|
|
)
|
|
return config
|
|
|
|
|
|
class _VisibleAgentStore:
|
|
async def get_visible(self, agent_id: str, user_id: str): # noqa: ARG002
|
|
return {"id": agent_id, "user_id": user_id}
|
|
|
|
async def ensure_builtin(self, record): # noqa: ARG002
|
|
return None
|
|
|
|
|
|
def _install_test_app(app: FastAPI, service, config) -> None:
|
|
app.state.config = config
|
|
app.state.maf_roundtable_service = service
|
|
app.state.maf_auto_dispatch = True
|
|
app.state.checkpointer = InMemorySaver()
|
|
app.state.thread_store = _ThreadStore()
|
|
app.state.run_manager = RunManager()
|
|
app.state.stream_bridge = MemoryStreamBridge()
|
|
app.state.run_event_store = MemoryRunEventStore()
|
|
app.state.store = None
|
|
app.state.agent_store = _VisibleAgentStore()
|
|
register_write_guard_service(service)
|
|
app.include_router(maf_router)
|
|
|
|
|
|
async def _wait_dispatch(dispatcher, run_id: str) -> None:
|
|
task = dispatcher._tasks.get(run_id) # noqa: SLF001 - entry test waits on the real task
|
|
assert task is not None
|
|
await task
|
|
|
|
|
|
@pytest.fixture
|
|
def deterministic_agent(monkeypatch, tmp_path):
|
|
agent_config = SimpleNamespace(
|
|
tool_groups=["maf-e2e"],
|
|
id="seat-1",
|
|
description="MAF E2E test seat",
|
|
model=None,
|
|
skills=[],
|
|
subagent_enabled=False,
|
|
name="MAF E2E seat",
|
|
)
|
|
paths = (
|
|
"deerflow.config.agents_config.load_agent_config",
|
|
"app.maf_roundtable.capability_snapshot.load_agent_config",
|
|
"deerflow.agents.lead_agent.agent.load_agent_config",
|
|
)
|
|
for path in paths:
|
|
monkeypatch.setattr(path, lambda _agent_id, cfg=agent_config: cfg)
|
|
monkeypatch.setattr("deerflow.agents.lead_agent.agent.load_agent_soul", lambda _agent_id: "")
|
|
monkeypatch.setattr(
|
|
"deerflow.agents.lead_agent.agent.create_chat_model",
|
|
lambda **_kwargs: _DeterministicModel(),
|
|
)
|
|
config = _app_config()
|
|
config.title.enabled = False
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "phase5-entry-test-secret")
|
|
monkeypatch.setenv("MAF_E2E_OUTBOX_DIR", str(tmp_path))
|
|
return config
|
|
|
|
|
|
async def _make_app(tmp_path, config):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_entry.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)
|
|
from app.maf_roundtable.service import MafRoundtableService
|
|
|
|
service = MafRoundtableService(
|
|
store,
|
|
enabled=True,
|
|
admin_diagnostic_access=False,
|
|
config=config.maf_roundtable,
|
|
agent_store=_VisibleAgentStore(),
|
|
app_config=config,
|
|
)
|
|
app = FastAPI()
|
|
_install_test_app(app, service, config)
|
|
|
|
@app.middleware("http")
|
|
async def test_auth(request: Request, call_next):
|
|
request.state.user = _User("u1")
|
|
return await call_next(request)
|
|
|
|
return app, service, engine
|
|
|
|
|
|
async def _create_waiting_run(client: AsyncClient, dispatcher, *, task_text: str, service=None) -> tuple[str, dict]:
|
|
session_response = await client.post(
|
|
"/api/maf-roundtable/sessions",
|
|
json={
|
|
"task_text": task_text,
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "seat-1", "role": "seat"}],
|
|
},
|
|
)
|
|
assert session_response.status_code == 200, session_response.text
|
|
session = session_response.json()
|
|
run_response = await client.post(
|
|
f"/api/maf-roundtable/sessions/{session['id']}/runs",
|
|
json={},
|
|
)
|
|
assert run_response.status_code == 200, run_response.text
|
|
run = run_response.json()
|
|
await _wait_dispatch(dispatcher, run["id"])
|
|
run_state = (await client.get(f"/api/maf-roundtable/runs/{run['id']}")).json()
|
|
if run_state["status"] != "waiting_approval" and service is not None:
|
|
actions = await service.store.actions.list_actions(run_id=run["id"])
|
|
raise AssertionError(f"unexpected run state: {run_state}; actions={actions}")
|
|
assert run_state["status"] == "waiting_approval"
|
|
return session["id"], run
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_entry_http_approval_reaches_real_toolnode_once(tmp_path, deterministic_agent):
|
|
app, service, engine = await _make_app(tmp_path, deterministic_agent)
|
|
dispatcher = get_dispatcher(app)
|
|
try:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
session_id, run = await _create_waiting_run(client, dispatcher, task_text="entry approval", service=service)
|
|
approvals = (await client.get(f"/api/maf-roundtable/runs/{run['id']}/approvals")).json()["approvals"]
|
|
assert len(approvals) == 1
|
|
approval_response = await client.post(
|
|
f"/api/maf-roundtable/approvals/{approvals[0]['id']}/decision",
|
|
json={"decision": "approved"},
|
|
)
|
|
assert approval_response.status_code == 200, approval_response.text
|
|
action = (await client.get(f"/api/maf-roundtable/runs/{run['id']}/actions")).json()["actions"][0]
|
|
assert action["status"] == "succeeded"
|
|
outbox = tmp_path / "phase5-target.jsonl"
|
|
assert outbox.exists()
|
|
assert outbox.read_text(encoding="utf-8").count("\n") == 1
|
|
assert (await client.get(f"/api/maf-roundtable/sessions/{session_id}")).json()["status"] == "completed"
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_entry_approval_scanner_expires_and_finishes_without_tool(tmp_path, deterministic_agent):
|
|
app, service, engine = await _make_app(tmp_path, deterministic_agent)
|
|
dispatcher = get_dispatcher(app)
|
|
try:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
_session_id, run = await _create_waiting_run(
|
|
client, dispatcher, task_text="scanner expiry", service=service
|
|
)
|
|
approval = (await client.get(
|
|
f"/api/maf-roundtable/runs/{run['id']}/approvals"
|
|
)).json()["approvals"][0]
|
|
async with service.store.session_factory() as db:
|
|
await db.execute(
|
|
update(MafRoundtableApprovalRow)
|
|
.where(MafRoundtableApprovalRow.id == approval["id"])
|
|
.values(expires_at=datetime.now(UTC) - timedelta(seconds=1))
|
|
)
|
|
await db.commit()
|
|
|
|
assert await _scan_maf_approval_expiry(app) == [run["id"]]
|
|
final_run = (await client.get(f"/api/maf-roundtable/runs/{run['id']}")).json()
|
|
assert final_run["status"] == "completed"
|
|
action = (await client.get(
|
|
f"/api/maf-roundtable/runs/{run['id']}/actions"
|
|
)).json()["actions"][0]
|
|
assert action["status"] == "rejected"
|
|
assert not (tmp_path / "phase5-target.jsonl").exists()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_entry_http_same_target_waits_for_explicit_resume(tmp_path, deterministic_agent):
|
|
app, service, engine = await _make_app(tmp_path, deterministic_agent)
|
|
dispatcher = get_dispatcher(app)
|
|
from deerflow.tools.builtins import maf_test_outbox
|
|
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
maf_test_outbox.configure_test_block(started=started, release=release)
|
|
try:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
_session_one, run_one = await _create_waiting_run(client, dispatcher, task_text="lease one", service=service)
|
|
approval_one = (await client.get(f"/api/maf-roundtable/runs/{run_one['id']}/approvals")).json()["approvals"][0]
|
|
first_decision = asyncio.create_task(client.post(
|
|
f"/api/maf-roundtable/approvals/{approval_one['id']}/decision",
|
|
json={"decision": "approved"},
|
|
))
|
|
assert await asyncio.to_thread(started.wait, 10)
|
|
|
|
_session_two, run_two = await _create_waiting_run(client, dispatcher, task_text="lease two", service=service)
|
|
approval_two = (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}/approvals")).json()["approvals"][0]
|
|
second_decision = await client.post(
|
|
f"/api/maf-roundtable/approvals/{approval_two['id']}/decision",
|
|
json={"decision": "approved"},
|
|
)
|
|
assert second_decision.status_code == 200
|
|
assert (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}")).json()["status"] == "waiting_approval"
|
|
second_action = (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}/actions")).json()["actions"][0]
|
|
assert second_action["status"] == "approved"
|
|
|
|
release.set()
|
|
assert (await first_decision).status_code == 200
|
|
assert (await client.get(f"/api/maf-roundtable/runs/{run_one['id']}")).json()["status"] == "completed"
|
|
|
|
resumed = await client.post(f"/api/maf-roundtable/runs/{run_two['id']}/resume")
|
|
assert resumed.status_code == 200, resumed.text
|
|
assert (await client.get(f"/api/maf-roundtable/runs/{run_two['id']}")).json()["status"] == "completed"
|
|
outbox = tmp_path / "phase5-target.jsonl"
|
|
assert outbox.read_text(encoding="utf-8").count("\n") == 2
|
|
finally:
|
|
release.set()
|
|
maf_test_outbox.clear_test_block()
|
|
await engine.dispose()
|