董宏凯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>
119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
"""P0-1: trusted capability snapshot builder tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from maf_test_helpers import FakeAgentStore
|
|
|
|
from app.maf_roundtable.capability_snapshot import (
|
|
AgentNotAccessibleError,
|
|
build_trusted_capability_snapshot,
|
|
)
|
|
from deerflow.config import get_app_config
|
|
from deerflow.tools.tool_metadata import resolve_tool_metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_forged_side_effect_not_stored(monkeypatch):
|
|
app_config = get_app_config()
|
|
|
|
def _fake_build_seat(*, app_config, agent_id, session_model_name, session_reasoning_effort, requested_excluded):
|
|
meta = resolve_tool_metadata("write_file")
|
|
return (["write_file"], [], [meta])
|
|
|
|
monkeypatch.setattr(
|
|
"app.maf_roundtable.capability_snapshot._seat_tools_for_agent",
|
|
_fake_build_seat,
|
|
)
|
|
|
|
snap = await build_trusted_capability_snapshot(
|
|
user_id="u1",
|
|
participants=[
|
|
{
|
|
"agent_id": "seat-a",
|
|
"role": "seat",
|
|
"excluded_tools": [],
|
|
# Client forgery ??must be ignored (field not read).
|
|
"capability_snapshot": {
|
|
"tools": [{"name": "write_file", "side_effect": "read", "risk_level": "R0"}],
|
|
},
|
|
}
|
|
],
|
|
app_config=app_config,
|
|
agent_store=FakeAgentStore(),
|
|
)
|
|
|
|
tool = snap["agents"][0]["tools"][0]
|
|
assert tool["side_effect"] == "write"
|
|
assert tool["risk_level"] == "R3"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_frozen_allow_list_excludes_registry_only_tool(monkeypatch):
|
|
app_config = get_app_config()
|
|
|
|
def _fake_build_seat(*, app_config, agent_id, session_model_name, session_reasoning_effort, requested_excluded):
|
|
return (["tool_a"], [], [resolve_tool_metadata("tool_a")])
|
|
|
|
monkeypatch.setattr(
|
|
"app.maf_roundtable.capability_snapshot._seat_tools_for_agent",
|
|
_fake_build_seat,
|
|
)
|
|
|
|
snap = await build_trusted_capability_snapshot(
|
|
user_id="u1",
|
|
participants=[{"agent_id": "seat-a", "role": "seat"}],
|
|
app_config=app_config,
|
|
agent_store=FakeAgentStore(),
|
|
)
|
|
|
|
assert snap["agents"][0]["allowed_tool_names"] == ["tool_a"]
|
|
assert "tool_b" not in snap["agents"][0]["allowed_tool_names"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inaccessible_agent_rejected():
|
|
app_config = get_app_config()
|
|
|
|
class _DenyStore:
|
|
async def get_visible(self, agent_id: str, user_id: str):
|
|
return None
|
|
|
|
with pytest.raises(AgentNotAccessibleError):
|
|
await build_trusted_capability_snapshot(
|
|
user_id="u1",
|
|
participants=[{"agent_id": "missing-agent", "role": "seat"}],
|
|
app_config=app_config,
|
|
agent_store=_DenyStore(),
|
|
)
|
|
|
|
|
|
def test_maf_tool_metadata_has_highest_server_priority():
|
|
metadata = resolve_tool_metadata(
|
|
"weknora-retrieval_weknora_search",
|
|
config_extra={"side_effect": "write", "risk_level": "R4"},
|
|
maf_metadata={"weknora-retrieval_weknora_search": {"side_effect": "read", "risk_level": "R0"}},
|
|
)
|
|
|
|
assert metadata["side_effect"] == "read"
|
|
assert metadata["risk_level"] == "R0"
|
|
|
|
|
|
def test_deployed_weknora_names_are_explicitly_read_only():
|
|
metadata = get_app_config().maf_roundtable.tool_metadata
|
|
|
|
assert metadata["weknora-retrieval_weknora_list_knowledge_bases"].side_effect == "read"
|
|
assert metadata["weknora-retrieval_weknora_list_knowledge_bases"].risk_level == "R0"
|
|
assert metadata["weknora-retrieval_weknora_search"].side_effect == "read"
|
|
assert metadata["weknora-retrieval_weknora_search"].risk_level == "R0"
|
|
|
|
|
|
def test_unknown_tool_metadata_remains_conservative():
|
|
metadata = resolve_tool_metadata(
|
|
"unregistered_external_tool",
|
|
config_extra={"side_effect": "invalid", "risk_level": "R99"},
|
|
)
|
|
|
|
assert metadata["side_effect"] == "write"
|
|
assert metadata["risk_level"] == "R3"
|