"""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, )