董宏凯andCursor
f82363d81f
- Backend: action_coordinator, capability_snapshot, event_normalizer, tool_executor updates - Backend: approved_tool_executor with write guard, action leases, migration script - Frontend: maf-roundtable module (API client, event timeline, approval cards, fact flow panel) - Frontend: MafRoutes, MafSessionsPage, MafRunDetailPage - Tests: action_leases, migration_phase5, phase5_entry_e2e, real_write_e2e/http_e2e - Docs: update implementation plan Co-authored-by: Cursor <cursoragent@cursor.com>
674 lines
23 KiB
Python
674 lines
23 KiB
Python
"""P0-1 acceptance tests for the harness approved-tool executor.
|
|
|
|
These exercise ``deerflow.tools.approved_tool_executor.execute_approved_tool``
|
|
directly on the real LangGraph ToolNode / ToolRuntime / middleware path with
|
|
controlled tools (no LLM, no DeerFlow runtime).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from langchain_core.tools import StructuredTool
|
|
from langgraph.prebuilt import ToolRuntime
|
|
|
|
from deerflow.agents.middlewares.tool_error_handling_middleware import (
|
|
build_lead_runtime_middlewares,
|
|
)
|
|
from deerflow.agents.middlewares.write_guard_middleware import (
|
|
WriteGuardMiddleware,
|
|
)
|
|
from deerflow.config.write_guard_config import WriteGuardConfig
|
|
from deerflow.tools import approved_tool_executor
|
|
from deerflow.tools.approved_tool_executor import execute_approved_tool
|
|
|
|
# ?? fakes ?????????????????????????????????????????????????????????????????
|
|
|
|
|
|
class _FakeCheckpointTuple:
|
|
def __init__(self, channel_values: dict[str, Any]) -> None:
|
|
self.checkpoint = {"channel_values": channel_values}
|
|
|
|
|
|
class _FakeCheckpointer:
|
|
def __init__(self, channel_values: dict[str, Any]) -> None:
|
|
self._cv = channel_values
|
|
|
|
async def aget_tuple(self, config: dict[str, Any]) -> _FakeCheckpointTuple:
|
|
return _FakeCheckpointTuple(self._cv)
|
|
|
|
|
|
CHANNEL_VALUES: dict[str, Any] = {
|
|
"thread_data": {"marker": "THREAD_MARKER_42"},
|
|
"sandbox": {"sandbox_id": "local"},
|
|
"messages": [],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def app_config():
|
|
from deerflow.config import get_app_config
|
|
|
|
# The production config mounts WriteGuard. These low-level executor tests
|
|
# that do not register a MAF service intentionally exercise the harness
|
|
# path without the app-layer validator; the dedicated enabled tests below
|
|
# provide the real bridge/service wiring.
|
|
return get_app_config().model_copy(update={"write_guard": WriteGuardConfig(enabled=False)})
|
|
|
|
|
|
def _snapshot(action_tool: str) -> dict[str, Any]:
|
|
meta = {"name": action_tool, "tool_name": action_tool, "side_effect": "write", "risk_level": "R3"}
|
|
return {
|
|
"version": 1,
|
|
"agents": [
|
|
{
|
|
"agent_id": "a1",
|
|
"allowed_tool_names": [action_tool],
|
|
"excluded_tools": [],
|
|
"tools": [meta],
|
|
}
|
|
],
|
|
"tools": [meta],
|
|
}
|
|
|
|
|
|
def _ctx(action_tool: str = "sync_write") -> dict[str, Any]:
|
|
"""Common trusted execution context + matching action record."""
|
|
return {
|
|
"thread_id": "thread-1",
|
|
"user_id": "u1",
|
|
"session_id": "s1",
|
|
"run_id": "r1",
|
|
"agent_id": "a1",
|
|
"turn_index": 2,
|
|
"model_name": None,
|
|
"excluded_tools": [],
|
|
"capability_snapshot": _snapshot(action_tool),
|
|
"write_policy": "confirm_high_risk",
|
|
"approved_action_id": "action-1",
|
|
"action_record": {
|
|
"id": "action-1",
|
|
"status": "running",
|
|
"run_id": "r1",
|
|
"session_id": "s1",
|
|
"agent_id": "a1",
|
|
"tool_name": action_tool,
|
|
"turn_index": 2,
|
|
},
|
|
}
|
|
|
|
|
|
# ?? 1. sync StructuredTool.func runs once ??succeeded ?????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_tool_runs_once_succeeded(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return f"sync-ok:{payload}"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="sync_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "succeeded", result
|
|
assert calls["n"] == 1, "sync tool must run exactly once"
|
|
|
|
|
|
# ?? 2. async tool runs normally ???????????????????????????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_tool_runs(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
async def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return f"async-ok:{payload}"
|
|
|
|
tool = StructuredTool.from_function(coroutine=_fn, name="async_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("async_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="async_write", params={"payload": "Y"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "succeeded", result
|
|
assert calls["n"] == 1
|
|
|
|
|
|
# ?? 3. tool needing ToolRuntime gets the REAL runtime + checkpoint state ??
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_receives_real_tool_runtime_and_state(app_config, monkeypatch):
|
|
seen: dict[str, Any] = {}
|
|
|
|
def _fn(payload: str, runtime: ToolRuntime[dict[str, Any], dict[str, Any]]) -> str:
|
|
# Typed ToolRuntime context avoids PydanticSerializationUnexpectedValue
|
|
# (bare ToolRuntime defaults context to None).
|
|
seen["runtime_type"] = type(runtime).__name__
|
|
seen["is_real_toolruntime"] = isinstance(runtime, ToolRuntime)
|
|
seen["marker"] = runtime.state.get("thread_data", {}).get("marker")
|
|
seen["context_run_id"] = runtime.context.get("maf_run_id")
|
|
seen["approved_tool"] = runtime.context.get("maf_approved_tool")
|
|
seen["context_action_id"] = runtime.context.get("maf_approved_action_id")
|
|
return "rt-ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="rt_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("rt_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="rt_write", params={"payload": "Z"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "succeeded", result
|
|
# A real ToolRuntime, not a duck-typed SimpleNamespace.
|
|
assert seen["is_real_toolruntime"] is True
|
|
assert seen["runtime_type"] != "SimpleNamespace"
|
|
# Real checkpoint state flowed through (sandbox / thread_data recovered).
|
|
assert seen["marker"] == "THREAD_MARKER_42"
|
|
# Trusted MAF context injected by the server (via Runtime on __pregel_runtime).
|
|
assert seen["context_run_id"] == "r1"
|
|
assert seen["approved_tool"] == "rt_write"
|
|
assert seen["context_action_id"] == "action-1"
|
|
|
|
|
|
# ?? 4. unauthorized / disabled / nonexistent ??0 executions ???????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonexistent_tool_executes_zero_times(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("ghost_tool") # action says ghost_tool; not in the allow-list
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="ghost_tool", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert calls["n"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_tool_executes_zero_times(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
ctx["capability_snapshot"]["agents"][0]["excluded_tools"] = ["sync_write"]
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="sync_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert calls["n"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_context_mismatch_executes_zero_times(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
# The stored action belongs to a DIFFERENT run ??must be refused.
|
|
ctx["action_record"] = {
|
|
"id": "action-1",
|
|
"status": "running",
|
|
"run_id": "OTHER_RUN",
|
|
"session_id": "s1",
|
|
"agent_id": "a1",
|
|
"tool_name": "sync_write",
|
|
"turn_index": 2,
|
|
}
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="sync_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert calls["n"] == 0
|
|
|
|
|
|
# ?? 5. LLM is never involved ??????????????????????????????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_llm_never_involved(app_config, monkeypatch):
|
|
llm_calls = {"n": 0}
|
|
|
|
def _boom(*args: Any, **kwargs: Any) -> Any:
|
|
llm_calls["n"] += 1
|
|
raise AssertionError("model factory must never be called by the approved executor")
|
|
|
|
# Guard every plausible model-construction entry point.
|
|
import deerflow.models.factory as factory
|
|
|
|
monkeypatch.setattr(factory, "create_chat_model", _boom)
|
|
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="sync_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "succeeded", result
|
|
assert calls["n"] == 1
|
|
assert llm_calls["n"] == 0, "the LLM must never be involved"
|
|
|
|
|
|
# ?? 6. inconclusive external errors ??unknown (no auto-retry) ???????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout_error_records_unknown(app_config, monkeypatch):
|
|
"""P1c: a timeout (external result unknowable) is recorded ``unknown`` ?? never ``failed`` and never auto-retried."""
|
|
|
|
async def _fn(payload: str) -> str:
|
|
raise TimeoutError("upstream timed out")
|
|
|
|
tool = StructuredTool.from_function(coroutine=_fn, name="timeout_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("timeout_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="timeout_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "unknown", result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_error_records_unknown(app_config, monkeypatch):
|
|
"""P1c: a dropped/reset connection is likewise inconclusive ??``unknown``."""
|
|
|
|
async def _fn(payload: str) -> str:
|
|
raise ConnectionResetError("connection reset by peer")
|
|
|
|
tool = StructuredTool.from_function(coroutine=_fn, name="conn_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("conn_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="conn_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "unknown", result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ordinary_error_records_failed(app_config, monkeypatch):
|
|
"""P1c: a deterministic validation error (external write definitely did NOT
|
|
happen) is ``failed``, not ``unknown``."""
|
|
|
|
async def _fn(payload: str) -> str:
|
|
raise ValueError("invalid payload value")
|
|
|
|
tool = StructuredTool.from_function(coroutine=_fn, name="bad_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("bad_write")
|
|
result = await execute_approved_tool(
|
|
app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
|
|
tool_name="bad_write", params={"payload": "X"}, **ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed", result
|
|
assert result.get("error") == "tool_execution_failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_error_never_leaks_secrets(app_config, monkeypatch, caplog):
|
|
secret = "TOP_SECRET_123"
|
|
|
|
def _fn(payload: str) -> str:
|
|
raise ValueError(f"token={secret}")
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="bad_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("bad_write")
|
|
with caplog.at_level("ERROR"):
|
|
result = await execute_approved_tool(
|
|
app_config=app_config,
|
|
checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
|
|
store=None,
|
|
tool_name="bad_write",
|
|
params={"payload": "X"},
|
|
**ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert secret not in str(result)
|
|
assert secret not in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_frozen_allow_list_blocks_extra_registry_tool(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _allowed(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
def _blocked(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "no"
|
|
|
|
allowed = StructuredTool.from_function(func=_allowed, name="tool_a", description="d")
|
|
blocked = StructuredTool.from_function(func=_blocked, name="tool_b", description="d")
|
|
monkeypatch.setattr(
|
|
approved_tool_executor, "get_available_tools", lambda **kw: [allowed, blocked]
|
|
)
|
|
|
|
snap = _snapshot("tool_a")
|
|
snap["agents"][0]["allowed_tool_names"] = ["tool_a"]
|
|
ctx = _ctx("tool_b")
|
|
ctx["capability_snapshot"] = snap
|
|
ctx["action_record"]["tool_name"] = "tool_b"
|
|
ctx["approved_action_id"] = "action-1"
|
|
|
|
result = await execute_approved_tool(
|
|
app_config=app_config,
|
|
checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
|
|
store=None,
|
|
tool_name="tool_b",
|
|
params={"payload": "X"},
|
|
**ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert calls["n"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_not_running_executes_zero_times(app_config, monkeypatch):
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
ctx["action_record"]["status"] = "approved"
|
|
result = await execute_approved_tool(
|
|
app_config=app_config,
|
|
checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
|
|
store=None,
|
|
tool_name="sync_write",
|
|
params={"payload": "X"},
|
|
**ctx,
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert calls["n"] == 0
|
|
|
|
|
|
def _enabled_write_guard_app_config(app_config):
|
|
enabled_guard = WriteGuardConfig.model_validate(
|
|
{
|
|
"enabled": True,
|
|
"provider": {
|
|
"policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
|
|
"interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
|
|
"config": {},
|
|
},
|
|
}
|
|
)
|
|
return app_config.model_copy(update={"write_guard": enabled_guard})
|
|
|
|
|
|
async def _writeguard_store_fixture(tmp_path, monkeypatch):
|
|
"""Create a real MAF store + service with WriteGuard bridge registered.
|
|
|
|
Uses monkeypatch so ``write_guard_bridge._maf_service`` is restored when the
|
|
test ends (avoids leaking a disposed sqlite service into later tests).
|
|
"""
|
|
from maf_test_helpers import make_maf_service
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
import deerflow.persistence.maf_roundtable.model # noqa: F401
|
|
from app.maf_roundtable import write_guard_bridge
|
|
from app.maf_roundtable.write_guard_bridge import register_write_guard_service
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'wg.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)
|
|
svc = make_maf_service(store)
|
|
# Capture the pre-test service *before* registration. Registering first
|
|
# would make monkeypatch restore this disposable sqlite-backed service at
|
|
# teardown and leak it into the next test.
|
|
monkeypatch.setattr(write_guard_bridge, "_maf_service", None)
|
|
register_write_guard_service(svc)
|
|
return engine, store, svc
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approved_tool_executor_with_writeguard_enabled(app_config, monkeypatch, tmp_path):
|
|
"""P1: real middleware chain with WriteGuard enabled passes approved execution."""
|
|
from app.maf_roundtable.write_guard_bridge import MafWriteGuardInterceptor
|
|
|
|
enabled_app_config = _enabled_write_guard_app_config(app_config)
|
|
middlewares = build_lead_runtime_middlewares(app_config=enabled_app_config)
|
|
assert any(isinstance(item, WriteGuardMiddleware) for item in middlewares)
|
|
|
|
engine, store, svc = await _writeguard_store_fixture(tmp_path, monkeypatch)
|
|
|
|
validate_spy_calls = {"n": 0}
|
|
original_validate = MafWriteGuardInterceptor.validate_approved_action
|
|
|
|
async def _spy_validate(self, *, tool_name: str, run_context: dict) -> bool:
|
|
validate_spy_calls["n"] += 1
|
|
return await original_validate(self, tool_name=tool_name, run_context=run_context)
|
|
|
|
monkeypatch.setattr(MafWriteGuardInterceptor, "validate_approved_action", _spy_validate)
|
|
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
action = await store.actions.create_action(
|
|
{
|
|
"session_id": session["id"],
|
|
"run_id": run["id"],
|
|
"agent_id": "a1",
|
|
"tool_name": "sync_write",
|
|
"idempotency_key": "wg-test-1",
|
|
"risk_level": "R3",
|
|
"status": "running",
|
|
"turn_index": 2,
|
|
}
|
|
)
|
|
|
|
calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
calls["n"] += 1
|
|
return "wg-ok"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
ctx = _ctx("sync_write")
|
|
ctx["session_id"] = session["id"]
|
|
ctx["run_id"] = run["id"]
|
|
ctx["approved_action_id"] = action["id"]
|
|
ctx["action_record"] = {
|
|
"id": action["id"],
|
|
"status": "running",
|
|
"run_id": run["id"],
|
|
"session_id": session["id"],
|
|
"agent_id": "a1",
|
|
"tool_name": "sync_write",
|
|
"turn_index": 2,
|
|
}
|
|
try:
|
|
result = await execute_approved_tool(
|
|
app_config=enabled_app_config,
|
|
checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
|
|
store=None,
|
|
tool_name="sync_write",
|
|
params={"payload": "X"},
|
|
**ctx,
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
assert result["status"] == "succeeded", result
|
|
assert calls["n"] == 1
|
|
assert validate_spy_calls["n"] == 1
|
|
assert result["result"] == {"executed": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"db_mismatch",
|
|
[
|
|
"status",
|
|
"run_id",
|
|
"session_id",
|
|
"agent_id",
|
|
"tool_name",
|
|
],
|
|
)
|
|
async def test_writeguard_rejects_db_mismatched_action(
|
|
app_config, monkeypatch, tmp_path, db_mismatch
|
|
):
|
|
"""Second-line defense: Guard re-reads the DB action and rejects mismatches.
|
|
|
|
``action_record`` passed to the executor stays consistent so the first-layer
|
|
field check passes; only the DB row is wrong.
|
|
"""
|
|
from app.maf_roundtable.write_guard_bridge import MafWriteGuardInterceptor
|
|
|
|
enabled_app_config = _enabled_write_guard_app_config(app_config)
|
|
middlewares = build_lead_runtime_middlewares(app_config=enabled_app_config)
|
|
assert any(isinstance(item, WriteGuardMiddleware) for item in middlewares)
|
|
|
|
engine, store, svc = await _writeguard_store_fixture(tmp_path, monkeypatch)
|
|
|
|
validate_spy_calls = {"n": 0}
|
|
original_validate = MafWriteGuardInterceptor.validate_approved_action
|
|
|
|
async def _spy_validate(self, *, tool_name: str, run_context: dict) -> bool:
|
|
validate_spy_calls["n"] += 1
|
|
return await original_validate(self, tool_name=tool_name, run_context=run_context)
|
|
|
|
monkeypatch.setattr(MafWriteGuardInterceptor, "validate_approved_action", _spy_validate)
|
|
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
other = await svc.create_session(user_id="u1", body={"task_text": "other"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
db_fields = {
|
|
"session_id": session["id"],
|
|
"run_id": run["id"],
|
|
"agent_id": "a1",
|
|
"tool_name": "sync_write",
|
|
"idempotency_key": f"wg-mismatch-{db_mismatch}",
|
|
"risk_level": "R3",
|
|
"status": "running",
|
|
"turn_index": 2,
|
|
}
|
|
if db_mismatch == "status":
|
|
db_fields["status"] = "approved"
|
|
elif db_mismatch == "run_id":
|
|
# No second create_run (session allows only one active run) ? fake id is enough.
|
|
db_fields["run_id"] = "other-run-id"
|
|
elif db_mismatch == "session_id":
|
|
db_fields["session_id"] = other["id"]
|
|
elif db_mismatch == "agent_id":
|
|
db_fields["agent_id"] = "other-agent"
|
|
elif db_mismatch == "tool_name":
|
|
db_fields["tool_name"] = "other_tool"
|
|
|
|
action = await store.actions.create_action(db_fields)
|
|
|
|
tool_calls = {"n": 0}
|
|
|
|
def _fn(payload: str) -> str:
|
|
tool_calls["n"] += 1
|
|
return "should-not-run"
|
|
|
|
tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
|
|
monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])
|
|
|
|
# Executor-facing record looks correct ? Guard must catch the DB mismatch.
|
|
ctx = _ctx("sync_write")
|
|
ctx["session_id"] = session["id"]
|
|
ctx["run_id"] = run["id"]
|
|
ctx["approved_action_id"] = action["id"]
|
|
ctx["action_record"] = {
|
|
"id": action["id"],
|
|
"status": "running",
|
|
"run_id": run["id"],
|
|
"session_id": session["id"],
|
|
"agent_id": "a1",
|
|
"tool_name": "sync_write",
|
|
"turn_index": 2,
|
|
}
|
|
try:
|
|
result = await execute_approved_tool(
|
|
app_config=enabled_app_config,
|
|
checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
|
|
store=None,
|
|
tool_name="sync_write",
|
|
params={"payload": "X"},
|
|
**ctx,
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
assert tool_calls["n"] == 0
|
|
assert validate_spy_calls["n"] == 1
|
|
assert result["status"] == "failed"
|