From 1a080311bdfb6e251b635faf4dc9f19f599be1bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 5 Aug 2026 21:47:46 +0800 Subject: [PATCH] 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 --- .../app/maf_roundtable/workflow_executors.py | 20 + .../app/maf_roundtable/workflow_runner.py | 96 ++++- .../maf_roundtable/participants.py | 139 +++++-- .../test_maf_phase3_dispatch_recovery.py | 379 ++++++++++++++++++ 4 files changed, 599 insertions(+), 35 deletions(-) create mode 100644 offline-backend-20260512/backend/tests/test_maf_phase3_dispatch_recovery.py diff --git a/offline-backend-20260512/backend/app/maf_roundtable/workflow_executors.py b/offline-backend-20260512/backend/app/maf_roundtable/workflow_executors.py index 7b13a364..a88d896e 100644 --- a/offline-backend-20260512/backend/app/maf_roundtable/workflow_executors.py +++ b/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. diff --git a/offline-backend-20260512/backend/app/maf_roundtable/workflow_runner.py b/offline-backend-20260512/backend/app/maf_roundtable/workflow_runner.py index 7f6cd438..c1e02182 100644 --- a/offline-backend-20260512/backend/app/maf_roundtable/workflow_runner.py +++ b/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": diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/participants.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/participants.py index 569fa0d8..cfafb02a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/participants.py +++ b/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: + 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: - existing_rows = ( + 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 diff --git a/offline-backend-20260512/backend/tests/test_maf_phase3_dispatch_recovery.py b/offline-backend-20260512/backend/tests/test_maf_phase3_dispatch_recovery.py new file mode 100644 index 00000000..63503e0f --- /dev/null +++ b/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()