fix(maf): harden phase-3 dispatch recovery and participant writes

Make ensure_participants concurrent-safe, restore clarification checkpoints without re-planning, and only create threads for validated seats.

Co-authored-by: Cursor <cursoragent@cursor.com>

offline-backend-20260512/backend/app/maf_roundtable/workflow_executors.py

@@ -648,6 +648,26 @@ async def _resolve_manager_dispatch(
         })
         await ctx.send_message({"decision": "clarification", "agents": []})
 
+    # ── P1-2: Clarification checkpoint recovery ──────────────────────────
+    # A clarification checkpoint stores a lightweight dict (not a full
+    # MafDispatchPlan).  Recovery must restore the original prompt without
+    # calling the manager model again, creating seat threads, or emitting
+    # duplicate audit events.
+    if state.intent_type == "clarification":
+        prompt = str((state.dispatch_plan or {}).get("clarification_prompt") or "").strip()
+        if not prompt:
+            await clarification("checkpoint_clarification_prompt_missing")
+            return
+
+        state.selected_seat_agent_ids = []
+        state.plan = None
+        state.status = "waiting_input"
+        state.stage = "waiting_input"
+        state.mark_completed(executor._node_id)
+        await _persist_dispatch_checkpoint(adapter, state)
+        await ctx.send_message({"decision": "clarification", "agents": [], "replayed": True})
+        return
+
     # A checkpoint written after validation survives process recovery and is
     # never re-planned. It still creates any selected threads that were not
     # bound before interruption.

offline-backend-20260512/backend/app/maf_roundtable/workflow_runner.py

@@ -95,7 +95,31 @@ class WorkflowRunner:
 
         try:
             participants = await self._service.store.participants.list_participants(session_id)
-            thread_map = await self._adapter.ensure_threads(session_id, run_id, participants)
+
+            task_payload = session.get("task_payload") or {}
+            maf_v2_payload = task_payload.get("maf_v2") if isinstance(task_payload, dict) else {}
+            maf_v2_payload = maf_v2_payload if isinstance(maf_v2_payload, dict) else {}
+            candidate_set = maf_v2_payload.get("candidate_set")
+            candidate_set = candidate_set if isinstance(candidate_set, dict) else {}
+
+            # P1-3: determine dispatch_contract_enabled before deciding which
+            # participants should receive threads on the first ensure_threads call.
+            dispatch_contract_enabled = bool(
+                maf_v2_payload.get("dispatch_contract_enabled")
+                or candidate_set
+                or maf_v2_payload.get("dispatch_plan") is not None
+            )
+
+            thread_map = await self._adapter.ensure_threads(
+                session_id,
+                run_id,
+                self._participants_requiring_threads(
+                    participants,
+                    dispatch_contract_enabled=dispatch_contract_enabled,
+                    dispatch_plan=maf_v2_payload.get("dispatch_plan"),
+                    intent_type=maf_v2_payload.get("intent_type"),
+                ),
+            )
 
             persisted_seat_agents = [
                 p["agent_id"] for p in participants if p.get("role") != "manager"
@@ -196,6 +220,62 @@ class WorkflowRunner:
         except Exception:  # noqa: BLE001
             return 600.0, True
 
+    def _participants_requiring_threads(
+        self,
+        participants: list[dict[str, Any]],
+        *,
+        dispatch_contract_enabled: bool,
+        dispatch_plan: dict[str, Any] | None,
+        intent_type: str | None,
+    ) -> list[dict[str, Any]]:
+        """Determine which participants should receive threads at startup.
+
+        P1-3: Old V2 sessions may have persisted ALL candidate seats before
+        dispatch was validated.  This method restricts the initial
+        ``ensure_threads`` call so that unvalidated seats never receive a
+        thread, enforcing the rule "only the manager gets a thread until the
+        dispatch plan is server-validated".
+
+        Rules:
+        ┌─────────────────────────────────────────────────────────────────────┐
+        │ State                        │ Allowed participants                 │
+        ├──────────────────────────────┼──────────────────────────────────────┤
+        │ New V2, not yet dispatched   │ manager only                         │
+        │ Validated direct_answer      │ manager only                         │
+        │ Validated clarification      │ manager only (usually no-op)         │
+        │ Validated normal dispatch    │ manager + selected seats from plan   │
+        │ Legacy MAF (no V2 contract)  │ all participants (original behavior) │
+        └──────────────────────────────┴──────────────────────────────────────┘
+        """
+        # Legacy MAF sessions without V2 contract keep original behavior.
+        if not dispatch_contract_enabled:
+            return participants
+
+        manager_participants = [
+            p for p in participants if p.get("role") == "manager"
+        ]
+
+        # No validated plan yet → only manager gets a thread.
+        if not dispatch_plan:
+            return manager_participants
+
+        plan_intent = str(dispatch_plan.get("intent_type") or intent_type or "")
+        execution_mode = str(dispatch_plan.get("execution_mode") or "")
+
+        # direct_answer or clarification → manager only.
+        if plan_intent in ("direct_answer", "clarification") or execution_mode == "manager_only":
+            return manager_participants
+
+        # Validated dispatch with selected seats → manager + those seats.
+        selected_ids = set(dispatch_plan.get("seat_agent_ids") or [])
+        if not selected_ids:
+            return manager_participants
+
+        return [
+            p for p in participants
+            if p.get("role") == "manager" or p.get("agent_id") in selected_ids
+        ]
+
     async def _run_manager_dispatch(self, state: RoundtableState) -> None:
         """Freeze the V2 manager decision before constructing seat topology."""
         if state.is_completed("manager_dispatch"):
@@ -774,9 +854,19 @@ class WorkflowRunner:
             )
             return await self._service.store.runs.get_run(run_id)
 
-        # Ensure threads still bound.
+        # Ensure threads still bound — but only for participants that are
+        # allowed to have threads at this stage (P1-3).
         participants = await self._service.store.participants.list_participants(state.session_id)
-        thread_map = await self._adapter.ensure_threads(state.session_id, run_id, participants)
+        thread_map = await self._adapter.ensure_threads(
+            state.session_id,
+            run_id,
+            self._participants_requiring_threads(
+                participants,
+                dispatch_contract_enabled=state.dispatch_contract_enabled,
+                dispatch_plan=state.dispatch_plan,
+                intent_type=state.intent_type,
+            ),
+        )
         state.thread_map.update(thread_map)
 
         if run["status"] == "waiting_approval":

offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/participants.py

@@ -9,15 +9,19 @@ creation time.
 
 from __future__ import annotations
 
+import logging
 from datetime import UTC, datetime
 from typing import Any
 from uuid import uuid4
 
 from sqlalchemy import delete, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.maf_roundtable.model import MafRoundtableParticipantRow
 
+logger = logging.getLogger(__name__)
+
 
 def _now() -> datetime:
     return datetime.now(UTC)
@@ -102,45 +106,116 @@ class MafParticipantRepository:
     async def ensure_participants(
         self, session_id: str, participants: list[dict[str, Any]]
     ) -> list[dict[str, Any]]:
-        """Insert missing participants without replacing existing threads.
+        """Insert missing participants using a database-level idempotent insert.
 
-        MAF Stage 3 initially persists only the manager participant.  After
-        the manager validates its dispatch plan, selected seats are added here
-        so unselected seats never receive a thread or execution slot.
+        Concurrent-safe: two coroutines calling this simultaneously with the
+        same seat set will never raise a unique-constraint error and will
+        never produce duplicate rows.
+
+        - SQLite / PostgreSQL: ``INSERT … ON CONFLICT (session_id, agent_id) DO NOTHING``
+        - MySQL: ``INSERT IGNORE …``
+
+        After the insert attempt the method **re-queries** the database and
+        returns the final persisted records (sorted by ``sort_order, agent_id``).
+        Existing ``thread_id``, ``sandbox_id``, ``status`` and
+        ``capability_snapshot_json`` are never overwritten.
         """
+        # Filter: skip empty agent_ids and de-duplicate input.
+        candidates: list[dict[str, Any]] = []
+        seen: set[str] = set()
+        for participant in participants:
+            agent_id = str(participant.get("agent_id") or "").strip()
+            if not agent_id or agent_id in seen:
+                continue
+            seen.add(agent_id)
+            candidates.append(participant)
+
+        if candidates:
             async with self._sf() as session:
-            existing_rows = (
+                await self._idempotent_insert(session, session_id, candidates)
+
+        # Re-query: always return the authoritative DB state.
+        return await self._list_sorted(session_id)
+
+    async def _idempotent_insert(
+        self,
+        session: AsyncSession,
+        session_id: str,
+        candidates: list[dict[str, Any]],
+    ) -> None:
+        """Execute a conflict-safe INSERT for the given candidate rows."""
+        from sqlalchemy import insert as sa_insert
+
+        now = _now()
+        rows = [
+            {
+                "id": uuid4().hex,
+                "session_id": session_id,
+                "agent_id": str(p.get("agent_id") or "").strip(),
+                "display_name": str(p.get("display_name") or ""),
+                "role": str(p.get("role") or "seat"),
+                "sort_order": int(p.get("sort_order", 0)),
+                "thread_id": None,
+                "sandbox_id": None,
+                "status": "idle",
+                "last_turn_index": None,
+                "capability_snapshot_json": p.get("capability_snapshot"),
+                "created_at": now,
+                "updated_at": now,
+            }
+            for p in candidates
+        ]
+
+        dialect_name = session.bind.dialect.name if session.bind else "sqlite"
+
+        if dialect_name in ("sqlite", "postgresql"):
+            # Use ON CONFLICT DO NOTHING for SQLite and PostgreSQL.
+            from sqlalchemy.dialects.postgresql import insert as pg_insert
+            from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+
+            dialect_insert = sqlite_insert if dialect_name == "sqlite" else pg_insert
+            stmt = dialect_insert(MafRoundtableParticipantRow).values(rows)
+            stmt = stmt.on_conflict_do_nothing(
+                index_elements=["session_id", "agent_id"],
+            )
+            await session.execute(stmt)
+            await session.commit()
+        elif dialect_name == "mysql":
+            # MySQL: INSERT IGNORE skips rows that violate a unique constraint.
+            from sqlalchemy.dialects.mysql import insert as mysql_insert
+
+            stmt = mysql_insert(MafRoundtableParticipantRow).values(rows)
+            stmt = stmt.prefix_with("IGNORE")
+            await session.execute(stmt)
+            await session.commit()
+        else:
+            # Fallback: try/except IntegrityError per row (rare dialects).
+            for row_data in rows:
+                try:
+                    session.add(MafRoundtableParticipantRow(**row_data))
+                    await session.commit()
+                except IntegrityError:
+                    await session.rollback()
+                    logger.debug(
+                        "ensure_participants: skipped duplicate (%s, %s)",
+                        session_id,
+                        row_data["agent_id"],
+                    )
+
+    async def _list_sorted(self, session_id: str) -> list[dict[str, Any]]:
+        """Return all participants for a session sorted by sort_order, agent_id."""
+        async with self._sf() as session:
+            rows = (
                 await session.execute(
-                    select(MafRoundtableParticipantRow).where(
-                        MafRoundtableParticipantRow.session_id == session_id
+                    select(MafRoundtableParticipantRow)
+                    .where(MafRoundtableParticipantRow.session_id == session_id)
+                    .order_by(
+                        MafRoundtableParticipantRow.sort_order.asc(),
+                        MafRoundtableParticipantRow.agent_id.asc(),
                     )
                 )
             ).scalars().all()
-            existing = {row.agent_id: row for row in existing_rows}
-            now = _now()
-            for participant in participants:
-                agent_id = str(participant.get("agent_id") or "").strip()
-                if not agent_id or agent_id in existing:
-                    continue
-                row = MafRoundtableParticipantRow(
-                    id=uuid4().hex,
-                    session_id=session_id,
-                    agent_id=agent_id,
-                    display_name=str(participant.get("display_name") or ""),
-                    role=str(participant.get("role") or "seat"),
-                    sort_order=int(participant.get("sort_order", 0)),
-                    thread_id=None,
-                    sandbox_id=None,
-                    status="idle",
-                    last_turn_index=None,
-                    capability_snapshot_json=participant.get("capability_snapshot"),
-                    created_at=now,
-                    updated_at=now,
-                )
-                session.add(row)
-                existing[agent_id] = row
-            await session.commit()
-            return [_to_dict(existing[agent_id]) for agent_id in existing]
+            return [_to_dict(r) for r in rows]
 
     async def update_participant_status(
         self, session_id: str, agent_id: str, *, status: str, last_turn_index: int | None = None

offline-backend-20260512/backend/tests/test_maf_phase3_dispatch_recovery.py

@@ -0,0 +1,379 @@
+"""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()