438 lines
16 KiB
Python
438 lines
16 KiB
Python
"""Deterministic Stage 3 tests for V2 conversation dispatch."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.conversation_service import MafConversationService
|
|
from app.maf_roundtable.dispatch_contracts import (
|
|
InvalidDispatchPlanError,
|
|
MafDispatchAssignment,
|
|
MafDispatchCandidateSet,
|
|
MafDispatchCatalog,
|
|
MafDispatchPlan,
|
|
MafIntentDecision,
|
|
resolve_intent,
|
|
validate_dispatch_plan,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableParticipantRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
|
|
class FakeMafConfig:
|
|
enabled = True
|
|
default_mode = "structured"
|
|
default_model_name = "test-model"
|
|
default_reasoning_effort = None
|
|
default_manager_agent_id = "manager-from-config"
|
|
default_seat_agent_ids = ["seat-a", "seat-b"]
|
|
business_routes = {}
|
|
max_seats_per_run = 4
|
|
default_max_rounds = 8
|
|
default_write_policy = "confirm_high_risk"
|
|
admin_diagnostic_access = True
|
|
|
|
|
|
class RecordingDispatcher:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[dict, dict, str]] = []
|
|
|
|
def check_sdk_available(self) -> None:
|
|
return
|
|
|
|
async def dispatch(self, session: dict, run: dict, *, user_id: str) -> None:
|
|
self.calls.append((session, run, user_id))
|
|
|
|
|
|
class CompletingDispatcher(RecordingDispatcher):
|
|
def __init__(self, store) -> None:
|
|
super().__init__()
|
|
self._store = store
|
|
self.ready = asyncio.Event()
|
|
|
|
async def dispatch(self, session: dict, run: dict, *, user_id: str) -> None:
|
|
await super().dispatch(session, run, user_id=user_id)
|
|
await self._store.runs.update_run(run["id"], {"status": "completed"})
|
|
await self._store.sessions.update_session(
|
|
session["id"], user_id=user_id, fields={"status": "completed"}
|
|
)
|
|
self.ready.set()
|
|
|
|
async def wait_for_run(self, run_id: str) -> None:
|
|
await self.ready.wait()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'phase3.db'}")
|
|
async with engine.begin() as conn:
|
|
import deerflow.persistence.maf_roundtable.model # noqa: F401
|
|
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield make_maf_roundtable_store(sf)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_route_creates_participants_and_dispatches(store):
|
|
dispatcher = RecordingDispatcher()
|
|
svc = MafConversationService(
|
|
store,
|
|
maf_config=FakeMafConfig(),
|
|
dispatcher=dispatcher,
|
|
)
|
|
conv = await svc.create_conversation(
|
|
user_id="u1",
|
|
title="Trusted route",
|
|
task_context={"business_code": "6BF"},
|
|
)
|
|
|
|
# The unknown business code is rejected rather than falling back across
|
|
# business boundaries.
|
|
with pytest.raises(Exception) as exc_info:
|
|
await svc.send_message(
|
|
conversation_id=conv["id"], user_id="u1", content="请多专家并行分析这个任务"
|
|
)
|
|
assert "route" in str(exc_info.value)
|
|
assert dispatcher.calls == []
|
|
|
|
# With no business code the configured default route is selected and the
|
|
# runtime sees only server-owned participants.
|
|
conv2 = await svc.create_conversation(user_id="u1", title="Default route")
|
|
await svc.send_message(
|
|
conversation_id=conv2["id"], user_id="u1", content="请多专家并行分析这个任务"
|
|
)
|
|
assert len(dispatcher.calls) == 1
|
|
session, run, owner = dispatcher.calls[0]
|
|
assert owner == "u1"
|
|
assert run["session_id"] == session["id"]
|
|
|
|
async with store.session_factory() as db:
|
|
session_row = (
|
|
await db.execute(
|
|
select(MafRoundtableSessionRow).where(
|
|
MafRoundtableSessionRow.id == session["id"]
|
|
)
|
|
)
|
|
).scalars().one()
|
|
participants = (
|
|
await db.execute(
|
|
select(MafRoundtableParticipantRow)
|
|
.where(MafRoundtableParticipantRow.session_id == session["id"])
|
|
.order_by(MafRoundtableParticipantRow.sort_order)
|
|
)
|
|
).scalars().all()
|
|
assert session_row.task_payload_json["maf_v2"]["manager_agent_id"] == "manager-from-config"
|
|
assert [p.agent_id for p in participants] == ["manager-from-config"]
|
|
assert all(p.thread_id is None for p in participants)
|
|
|
|
events = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv2["id"]
|
|
)
|
|
)
|
|
).scalars().one()
|
|
assert events.active_run_id == run["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_answer_persists_manager_only_participants(store):
|
|
dispatcher = RecordingDispatcher()
|
|
svc = MafConversationService(store, maf_config=FakeMafConfig(), dispatcher=dispatcher)
|
|
conv = await svc.create_conversation(user_id="u1", title="Manager only")
|
|
|
|
await svc.send_message(
|
|
conversation_id=conv["id"],
|
|
user_id="u1",
|
|
content="给我一个普通的直接答案",
|
|
)
|
|
assert len(dispatcher.calls) == 1
|
|
session, _run, _owner = dispatcher.calls[0]
|
|
|
|
async with store.session_factory() as db:
|
|
participants = (
|
|
await db.execute(
|
|
select(MafRoundtableParticipantRow)
|
|
.where(MafRoundtableParticipantRow.session_id == session["id"])
|
|
.order_by(MafRoundtableParticipantRow.sort_order)
|
|
)
|
|
).scalars().all()
|
|
assert [participant.agent_id for participant in participants] == [
|
|
"manager-from-config"
|
|
]
|
|
maf_v2 = session["task_payload"]["maf_v2"]
|
|
assert maf_v2["manager_agent_id"] == "manager-from-config"
|
|
assert maf_v2["candidate_seat_agent_ids"] == ["seat-a", "seat-b"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clarification_does_not_create_or_dispatch_run(store):
|
|
"""Short / unclear content still creates a run — the manager workflow
|
|
is responsible for detecting clarification needs, not the send path."""
|
|
dispatcher = RecordingDispatcher()
|
|
svc = MafConversationService(store, maf_config=FakeMafConfig(), dispatcher=dispatcher)
|
|
conv = await svc.create_conversation(user_id="u1", title="Need detail")
|
|
|
|
result = await svc.send_message(conversation_id=conv["id"], user_id="u1", content="请分析这个任务")
|
|
assert result["message_id"]
|
|
assert len(dispatcher.calls) == 1
|
|
|
|
async with store.session_factory() as db:
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
)
|
|
)
|
|
).scalars().one()
|
|
assert conv_row.safe_status == "running"
|
|
assert conv_row.active_run_id is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatcher_completion_bridges_back_to_conversation(store):
|
|
dispatcher = CompletingDispatcher(store)
|
|
svc = MafConversationService(store, maf_config=FakeMafConfig(), dispatcher=dispatcher)
|
|
conv = await svc.create_conversation(user_id="u1", title="Completion")
|
|
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="complete this task")
|
|
watcher_tasks = list(svc._watchers.values()) # noqa: SLF001 - lifecycle assertion
|
|
if watcher_tasks:
|
|
await asyncio.gather(*watcher_tasks)
|
|
|
|
public = await svc.get_conversation(conversation_id=conv["id"], user_id="u1")
|
|
assert public["safe_status"] == "completed"
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert [event["type"] for event in events][-1] == "run.completed"
|
|
assert all(event["run_id"] is None for event in events)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workflow_projection_whitelists_user_visible_event_data(store):
|
|
svc = MafConversationService(store, maf_config=FakeMafConfig())
|
|
conv = await svc.create_conversation(user_id="u1", title="Projection")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="complete this task")
|
|
internal = await store.conversations.get_conversation(conv["id"], user_id="u1")
|
|
run = await store.runs.list_runs(internal["internal_session_id"])
|
|
|
|
await svc.bridge_workflow_event(
|
|
run_id=run[0]["id"],
|
|
event={
|
|
"event_type": "tool.requested",
|
|
"agent_id": "seat-a",
|
|
"turn_index": 1,
|
|
"payload": {
|
|
"tool_name": "safe_tool",
|
|
"args": {"secret": "must-not-cross"},
|
|
"tool_call_id": "internal-call-id",
|
|
},
|
|
},
|
|
)
|
|
await svc.bridge_workflow_event(
|
|
run_id=run[0]["id"],
|
|
event={
|
|
"event_type": "agent.message.completed",
|
|
"agent_id": "seat-a",
|
|
"turn_index": 1,
|
|
"payload": {
|
|
"content": "answer",
|
|
"metadata": {"thread_id": "thread-secret", "deerflow_run_id": "df-secret"},
|
|
},
|
|
},
|
|
)
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=1)
|
|
tool = next(event for event in events if event["type"] == "tool.requested")
|
|
assert tool["data"] == {"tool_name": "safe_tool", "agent_id": "seat-a", "turn_index": 1}
|
|
complete = next(event for event in events if event["type"] == "agent.message.complete")
|
|
assert "metadata" not in complete["data"]
|
|
assert "thread-secret" not in str(complete)
|
|
assert any(event["type"] == "seat.delivery.submitted" for event in events)
|
|
|
|
|
|
def test_intent_resolution_is_deterministic_and_does_not_select_agents():
|
|
assert resolve_intent("请多专家并行分析") .intent_type == "parallel_consultation"
|
|
assert resolve_intent("请发布这条内容").intent_type == "tool_write"
|
|
assert resolve_intent("不清楚").intent_type == "clarification"
|
|
assert resolve_intent("给我一个普通答案").intent_type == "direct_answer"
|
|
|
|
|
|
def test_new_conversation_candidate_set_has_no_keyword_intent_decision():
|
|
catalog = MafDispatchCatalog(FakeMafConfig())
|
|
candidate_set = catalog.build_candidate_set(task_context=None)
|
|
|
|
assert candidate_set == MafDispatchCandidateSet(
|
|
manager_agent_id="manager-from-config",
|
|
candidate_seat_agent_ids=["seat-a", "seat-b"],
|
|
mode="structured",
|
|
max_rounds=8,
|
|
)
|
|
# The request wording cannot change this server-owned candidate set;
|
|
# the manager JSON contract decides intent only after this point.
|
|
assert catalog.build_candidate_set(task_context=None) == candidate_set
|
|
|
|
|
|
def test_weak_model_cannot_expand_or_duplicate_dispatch_plan():
|
|
with pytest.raises(ValueError):
|
|
MafDispatchPlan(
|
|
manager_agent_id="manager",
|
|
seat_agent_ids=["seat-a", "seat-a"],
|
|
assignments=[
|
|
MafDispatchAssignment(agent_id="seat-a", sort_order=1),
|
|
MafDispatchAssignment(agent_id="seat-a", sort_order=2),
|
|
],
|
|
intent_type="parallel_consultation",
|
|
max_rounds=8,
|
|
)
|
|
|
|
plan = MafDispatchPlan(
|
|
manager_agent_id="manager",
|
|
seat_agent_ids=["seat-a"],
|
|
assignments=[MafDispatchAssignment(agent_id="seat-a", sort_order=1)],
|
|
intent_type="single_seat",
|
|
max_rounds=8,
|
|
execution_mode="single",
|
|
)
|
|
with pytest.raises(InvalidDispatchPlanError):
|
|
validate_dispatch_plan(plan, trusted_agent_ids={"manager", "seat-b"}, max_seats=4)
|
|
|
|
with pytest.raises(ValueError, match="at least one configured seat"):
|
|
MafDispatchPlan(
|
|
manager_agent_id="manager",
|
|
seat_agent_ids=[],
|
|
assignments=[],
|
|
intent_type="parallel_consultation",
|
|
max_rounds=8,
|
|
)
|
|
|
|
|
|
def test_clarification_plan_is_not_constructible():
|
|
with pytest.raises(ValueError):
|
|
MafIntentDecision(
|
|
intent_type="clarification",
|
|
confidence=0.9,
|
|
reason_code="needs_detail",
|
|
)
|
|
|
|
|
|
def test_server_dispatch_plan_selects_the_correct_topology():
|
|
catalog = MafDispatchCatalog(FakeMafConfig())
|
|
cases = [
|
|
("direct_answer", [], "manager_only"),
|
|
("single_seat", ["seat-a"], "single"),
|
|
("parallel_consultation", ["seat-a", "seat-b"], "parallel"),
|
|
("chain_consultation", ["seat-a", "seat-b"], "chain"),
|
|
("tool_write", ["seat-a"], "single"),
|
|
]
|
|
for intent_name, expected_seats, expected_mode in cases:
|
|
plan = catalog.build_plan(
|
|
intent=MafIntentDecision(
|
|
intent_type=intent_name,
|
|
confidence=1,
|
|
reason_code="test",
|
|
),
|
|
task_context=None,
|
|
)
|
|
assert plan.seat_agent_ids == expected_seats
|
|
assert plan.execution_mode == expected_mode
|
|
assert plan.candidate_seat_agent_ids == ["seat-a", "seat-b"]
|
|
|
|
single_plan = catalog.build_plan(
|
|
intent=MafIntentDecision(
|
|
intent_type="single_seat",
|
|
confidence=1,
|
|
reason_code="test",
|
|
),
|
|
task_context=None,
|
|
)
|
|
participant_ids = [item["agent_id"] for item in catalog.participants(single_plan)]
|
|
assert participant_ids == ["manager-from-config", "seat-a", "seat-b"]
|
|
|
|
|
|
def test_manager_decision_selects_intent_only_within_trusted_candidates():
|
|
candidate_set = MafDispatchCatalog(FakeMafConfig()).build_candidate_set(task_context=None)
|
|
proposed, plan = MafDispatchCatalog.parse_manager_decision(
|
|
json.dumps({
|
|
"intent_type": "direct_answer",
|
|
"selected_seat_agent_ids": [],
|
|
"assignments": [],
|
|
"execution_mode": "manager_only",
|
|
"convergence_path": ["manager answer"],
|
|
"reason": "manager can answer directly",
|
|
}),
|
|
candidate_set=candidate_set,
|
|
max_seats=4,
|
|
)
|
|
assert proposed.intent_type == "direct_answer"
|
|
assert plan is not None
|
|
assert plan.execution_mode == "manager_only"
|
|
assert plan.seat_agent_ids == []
|
|
|
|
|
|
def test_manager_plan_is_schema_checked_and_allowlisted():
|
|
catalog = MafDispatchCatalog(FakeMafConfig())
|
|
server_plan = catalog.build_plan(
|
|
intent=MafIntentDecision(
|
|
intent_type="parallel_consultation",
|
|
confidence=1,
|
|
reason_code="test",
|
|
),
|
|
task_context=None,
|
|
)
|
|
raw = json.dumps({
|
|
"intent_type": "parallel_consultation",
|
|
"selected_seat_agent_ids": ["seat-a", "seat-b"],
|
|
"assignments": [{
|
|
"agent_id": "seat-a",
|
|
"task_goal": "Collect the first evidence set",
|
|
"delivery_requirements": ["cite evidence"],
|
|
"order": 1,
|
|
}, {
|
|
"agent_id": "seat-b",
|
|
"task_goal": "Compare the two evidence sets",
|
|
"delivery_requirements": ["cite evidence"],
|
|
"order": 2,
|
|
}],
|
|
"execution_mode": "parallel",
|
|
"convergence_path": ["manager synthesis", "cross review", "final answer"],
|
|
"reason": "seat-b has the relevant capability",
|
|
})
|
|
final_plan = catalog.parse_manager_plan(
|
|
raw,
|
|
expected_intent="parallel_consultation",
|
|
server_plan=server_plan,
|
|
max_seats=4,
|
|
)
|
|
assert final_plan.seat_agent_ids == ["seat-a", "seat-b"]
|
|
assert final_plan.source == "manager"
|
|
|
|
for invalid in (
|
|
raw.replace('"seat-b"', '"untrusted-seat"'),
|
|
raw.replace('"selected_seat_agent_ids": ["seat-a", "seat-b"]', '"selected_seat_agent_ids": ["seat-b", "seat-b"]'),
|
|
raw.replace('"execution_mode": "parallel"', '"execution_mode": "single"'),
|
|
raw.replace('"delivery_requirements": ["cite evidence"]', '"delivery_requirements": []'),
|
|
raw.replace(
|
|
'"convergence_path": ["manager synthesis", "cross review", "final answer"]',
|
|
'"convergence_path": []',
|
|
),
|
|
):
|
|
with pytest.raises(InvalidDispatchPlanError):
|
|
catalog.parse_manager_plan(
|
|
invalid,
|
|
expected_intent="parallel_consultation",
|
|
server_plan=server_plan,
|
|
max_seats=4,
|
|
)
|