"""P1-1 / P1-2: concurrent-safe ensure_participants + clarification checkpoint recovery. P1-1: Two coroutines calling ensure_participants() concurrently must not raise unique-constraint errors or produce duplicate rows. P1-2: A clarification checkpoint must restore the original prompt without calling the manager model again, creating seat threads, or emitting duplicate audit events. """ from __future__ import annotations import asyncio import json from datetime import datetime from types import SimpleNamespace from unittest.mock import AsyncMock import pytest import pytest_asyncio from maf_test_helpers import make_maf_service from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.maf_roundtable.workflow_executors import ( ManagerDispatchExecutor, RoundtableState, ) from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.persistence.maf_roundtable.model import ( MafRoundtableParticipantRow, MafRoundtableRunRow, MafRoundtableSessionRow, ) # ── Fixtures ───────────────────────────────────────────────────────────────── @pytest_asyncio.fixture async def store(tmp_path): """File-based SQLite store for concurrency tests (in-memory shares one connection across sessions, making true races impossible).""" engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'p1.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) store = make_maf_roundtable_store(sf) yield store await engine.dispose() # ── P1-1: Concurrent ensure_participants ───────────────────────────────────── @pytest.mark.asyncio async def test_concurrent_ensure_participants_no_duplicates(store): """Two coroutines calling ensure_participants() with the same seat set must both succeed without raising and without producing duplicate rows.""" # Seed a session with a manager participant. session = await store.sessions.create_session({ "user_id": "u1", "task_text": "test", "mode": "structured", "manager_agent_id": "manager", "max_rounds": 8, }) await store.participants.replace_participants(session["id"], [{ "agent_id": "manager", "role": "manager", "sort_order": 0, }]) seat_rows = [ {"agent_id": "seat-a", "role": "seat", "sort_order": 1, "capability_snapshot": {"tools": []}}, {"agent_id": "seat-b", "role": "seat", "sort_order": 2, "capability_snapshot": {"tools": []}}, ] errors: list[Exception] = [] async def _ensure(): try: await store.participants.ensure_participants(session["id"], seat_rows) except Exception as exc: # noqa: BLE001 errors.append(exc) # Run two concurrent ensures. await asyncio.gather(_ensure(), _ensure()) assert errors == [], f"ensure_participants raised: {errors}" # Verify final DB state: exactly manager, seat-a, seat-b — one each. participants = await store.participants.list_participants(session["id"]) agent_ids = [p["agent_id"] for p in participants] assert sorted(agent_ids) == ["manager", "seat-a", "seat-b"] # No duplicates. assert len(agent_ids) == len(set(agent_ids)) @pytest.mark.asyncio async def test_ensure_participants_preserves_existing_thread_id(store): """A seat that already has a bound thread_id must NOT have it cleared or overwritten by a subsequent ensure call.""" session = await store.sessions.create_session({ "user_id": "u1", "task_text": "test", "mode": "structured", "manager_agent_id": "manager", "max_rounds": 8, }) await store.participants.replace_participants(session["id"], [{ "agent_id": "manager", "role": "manager", "sort_order": 0, }]) # Insert seat-a with a bound thread. await store.participants.ensure_participants(session["id"], [ {"agent_id": "seat-a", "role": "seat", "sort_order": 1}, ]) await store.participants.bind_thread(session["id"], "seat-a", thread_id="thread-bound-a") # Now ensure again with the same seat (simulating a concurrent recovery). await store.participants.ensure_participants(session["id"], [ {"agent_id": "seat-a", "role": "seat", "sort_order": 1}, ]) # thread_id must be preserved. participants = await store.participants.list_participants(session["id"]) seat_a = next(p for p in participants if p["agent_id"] == "seat-a") assert seat_a["thread_id"] == "thread-bound-a" @pytest.mark.asyncio async def test_ensure_participants_skips_empty_agent_id(store): """Participants with empty or missing agent_id are skipped.""" session = await store.sessions.create_session({ "user_id": "u1", "task_text": "test", "mode": "structured", "manager_agent_id": "manager", "max_rounds": 8, }) await store.participants.replace_participants(session["id"], [{ "agent_id": "manager", "role": "manager", "sort_order": 0, }]) await store.participants.ensure_participants(session["id"], [ {"agent_id": "", "role": "seat"}, {"agent_id": None, "role": "seat"}, {"role": "seat"}, # no agent_id key {"agent_id": "seat-valid", "role": "seat", "sort_order": 1}, ]) participants = await store.participants.list_participants(session["id"]) agent_ids = [p["agent_id"] for p in participants] assert "seat-valid" in agent_ids assert "" not in agent_ids assert None not in agent_ids @pytest.mark.asyncio async def test_ensure_participants_sorted_by_sort_order_and_agent_id(store): """Results are sorted by (sort_order, agent_id).""" session = await store.sessions.create_session({ "user_id": "u1", "task_text": "test", "mode": "structured", "manager_agent_id": "manager", "max_rounds": 8, }) await store.participants.replace_participants(session["id"], [{ "agent_id": "manager", "role": "manager", "sort_order": 0, }]) await store.participants.ensure_participants(session["id"], [ {"agent_id": "seat-z", "role": "seat", "sort_order": 2}, {"agent_id": "seat-a", "role": "seat", "sort_order": 1}, {"agent_id": "seat-m", "role": "seat", "sort_order": 1}, ]) participants = await store.participants.list_participants(session["id"]) agent_ids = [p["agent_id"] for p in participants] # manager (sort_order=0) first, then sort_order=1 (seat-a, seat-m), then sort_order=2 (seat-z) assert agent_ids == ["manager", "seat-a", "seat-m", "seat-z"] # ── P1-2: Clarification checkpoint recovery ────────────────────────────────── class _FakeCtx: def __init__(self) -> None: self.messages: list[dict] = [] self.outputs: list[dict] = [] async def send_message(self, msg: dict) -> None: self.messages.append(msg) async def yield_output(self, msg: dict) -> None: self.outputs.append(msg) @pytest.mark.asyncio async def test_clarification_checkpoint_recovery_preserves_prompt(): """A clarification checkpoint must restore the original prompt without calling the manager model again or creating seat threads.""" original_prompt = "请明确您的业务目标和预期产出。" # Build a state as if the manager returned clarification. state = RoundtableState( session_id="s1", run_id="r1", task_text="分析市场", mode="structured", model_name=None, reasoning_effort=None, max_rounds=8, write_policy="confirm_high_risk", manager_agent_id="manager", seat_agent_ids=["seat-a", "seat-b"], thread_map={"manager": "t-manager"}, capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True, intent_type="clarification", dispatch_plan={ "execution_mode": "manager_only", "convergence_path": ["clarification"], "reason": "insufficient_task_detail", "source": "manager", "clarification_prompt": original_prompt, }, dispatch_contract_enabled=True, ) state.selected_seat_agent_ids = [] state.status = "waiting_input" state.stage = "waiting_input" state.mark_completed("manager_dispatch") # Serialize and restore (simulates process restart). checkpoint = state.to_dict() restored = RoundtableState.from_dict(checkpoint) # The manager model must NOT be called again. adapter = SimpleNamespace( run_seat_turn=AsyncMock(side_effect=AssertionError("manager must not be called on recovery")), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace( runs=SimpleNamespace(update_run=AsyncMock()), )), ) ctx = _FakeCtx() executor = ManagerDispatchExecutor(adapter, restored) await executor.handle("task", ctx) # Assertions: # 1. Original clarification prompt is preserved. assert restored.dispatch_plan["clarification_prompt"] == original_prompt # 2. Manager model was NOT called. adapter.run_seat_turn.assert_not_awaited() # 3. No seat threads created. assert restored.selected_seat_agent_ids == [] # 4. Status is waiting_input. assert restored.status == "waiting_input" assert restored.stage == "waiting_input" # 5. The message sent is a replay, not a new dispatch. assert len(ctx.messages) == 1 assert ctx.messages[0]["decision"] == "clarification" assert ctx.messages[0].get("replayed") is True @pytest.mark.asyncio async def test_clarification_checkpoint_without_prompt_falls_back(): """If the checkpoint has no clarification_prompt, the executor falls back to the unified safe clarification.""" state = RoundtableState( session_id="s1", run_id="r1", task_text="分析市场", mode="structured", model_name=None, reasoning_effort=None, max_rounds=8, write_policy="confirm_high_risk", manager_agent_id="manager", seat_agent_ids=[], thread_map={"manager": "t-manager"}, capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True, intent_type="clarification", dispatch_plan={ "execution_mode": "manager_only", "convergence_path": ["clarification"], "reason": "some_reason", "source": "manager", # No clarification_prompt key! }, dispatch_contract_enabled=True, ) state.mark_completed("manager_dispatch") # Serialize and restore. checkpoint = state.to_dict() restored = RoundtableState.from_dict(checkpoint) adapter = SimpleNamespace( run_seat_turn=AsyncMock(), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace( runs=SimpleNamespace(update_run=AsyncMock()), )), ) ctx = _FakeCtx() executor = ManagerDispatchExecutor(adapter, restored) await executor.handle("task", ctx) # Should fall back to safe clarification with a generated prompt. assert restored.status == "waiting_input" assert restored.dispatch_plan.get("clarification_prompt") is not None assert restored.dispatch_plan["reason"] == "checkpoint_clarification_prompt_missing" @pytest.mark.asyncio async def test_clarification_recovery_does_not_emit_duplicate_events(): """Recovery of a clarification checkpoint must not emit duplicate clarification.required audit events.""" original_prompt = "请说明具体需求。" state = RoundtableState( session_id="s1", run_id="r1", task_text="t", mode="structured", model_name=None, reasoning_effort=None, max_rounds=8, write_policy="confirm_high_risk", manager_agent_id="manager", seat_agent_ids=[], thread_map={"manager": "t-manager"}, capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True, intent_type="clarification", dispatch_plan={ "execution_mode": "manager_only", "convergence_path": ["clarification"], "reason": "insufficient_task_detail", "source": "manager", "clarification_prompt": original_prompt, }, dispatch_contract_enabled=True, ) state.mark_completed("manager_dispatch") checkpoint = state.to_dict() restored = RoundtableState.from_dict(checkpoint) adapter = SimpleNamespace( run_seat_turn=AsyncMock(), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace( runs=SimpleNamespace(update_run=AsyncMock()), )), ) ctx = _FakeCtx() executor = ManagerDispatchExecutor(adapter, restored) await executor.handle("task", ctx) # _emit_critical should NOT be called during recovery (no duplicate events). adapter._emit_critical.assert_not_awaited()