fix(maf): recover clarification checkpoints stuck in running

Re-enter clarification recovery even when manager_dispatch is already completed, and expand V2 runner tests for group_chat seat filtering and recover().

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

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

@@ -358,6 +358,15 @@ class ManagerDispatchExecutor(Executor):
 
     @handler(input=Any, output=dict)
     async def handle(self, message: Any, ctx: WorkflowContext) -> None:
+        # V2 clarification checkpoints must always re-enter the clarification
+        # recovery branch, even when manager_dispatch is already marked
+        # completed — otherwise a recovered clarification run would stay
+        # "running" instead of landing in waiting_input.
+        if self._state.dispatch_contract_enabled and self._state.intent_type == "clarification":
+            self._state.stage = "planning"
+            await _resolve_manager_dispatch(self, ctx)
+            return
+
         if self._state.is_completed(self._node_id):
             await ctx.send_message({
                 "decision": self._dispatch_decision(),

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

@@ -278,7 +278,14 @@ class WorkflowRunner:
 
     async def _run_manager_dispatch(self, state: RoundtableState) -> None:
         """Freeze the V2 manager decision before constructing seat topology."""
-        if state.is_completed("manager_dispatch"):
+        # A completed dispatch is replayed without another model call — except
+        # for clarification checkpoints, which must always re-enter the
+        # clarification recovery branch so a recovered run lands in
+        # waiting_input instead of staying "running" forever.
+        if (
+            state.is_completed("manager_dispatch")
+            and state.intent_type != "clarification"
+        ):
             return
         workflow = build_manager_dispatch_workflow(
             manager_dispatch=ManagerDispatchExecutor(self._adapter, state),

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

@@ -144,8 +144,6 @@ class MafParticipantRepository:
         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 = [
             {

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

@@ -11,14 +11,11 @@ P1-2: A clarification checkpoint must restore the original prompt without
 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 (
@@ -27,12 +24,6 @@ from app.maf_roundtable.workflow_executors import (
 )
 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 ─────────────────────────────────────────────────────────────────
 

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

@@ -38,7 +38,12 @@ from app.maf_roundtable.workflow_executors import (
     SeatExecutor,
 )
 from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
-from deerflow.persistence.maf_roundtable.model import Base, MafRoundtableRunRow, MafRoundtableSessionRow
+from deerflow.persistence.maf_roundtable.model import (
+    Base,
+    MafRoundtableParticipantRow,
+    MafRoundtableRunRow,
+    MafRoundtableSessionRow,
+)
 
 MANAGER = "manager-1"
 SEATS = ["seat-a", "seat-b"]
@@ -271,49 +276,53 @@ async def test_manager_role_wired_through_state(svc):
 async def v2_svc(tmp_path):
     """Service with a V2 session: manager + 3 candidate seats, dispatch_contract_enabled."""
     engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'v2runner.db'}", echo=False)
-    async with engine.begin() as conn:
+    try:
-        await conn.run_sync(Base.metadata.create_all)
+        async with engine.begin() as conn:
-    sf = async_sessionmaker(engine, expire_on_commit=False)
+            await conn.run_sync(Base.metadata.create_all)
-    store = make_maf_roundtable_store(sf)
+        sf = async_sessionmaker(engine, expire_on_commit=False)
-    service = make_maf_service(store)
+        store = make_maf_roundtable_store(sf)
-    now = datetime(2026, 1, 1)
+        service = make_maf_service(store)
-    async with sf() as session:
+        now = datetime(2026, 1, 1)
-        session.add(MafRoundtableSessionRow(
+        async with sf() as session:
-            id="v2-sess", user_id="u1", title="v2", task_text="分析市场趋势",
+            session.add(MafRoundtableSessionRow(
-            mode="structured", status="pending", write_policy="auto",
+                id="v2-sess", user_id="u1", title="v2", task_text="分析市场趋势",
-            manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
+                mode="structured", status="pending", write_policy="auto",
-            task_payload=json.dumps({
+                manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
-                "maf_v2": {
+                task_payload_json={
-                    "dispatch_contract_enabled": True,
+                    "maf_v2": {
-                    "candidate_set": {
+                        "dispatch_contract_enabled": True,
-                        "manager_agent_id": MANAGER,
+                        "candidate_set": {
-                        "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"],
+                            "manager_agent_id": MANAGER,
-                        "mode": "structured",
+                            "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"],
-                        "max_rounds": 8,
+                            "mode": "structured",
+                            "max_rounds": 8,
+                        },
                     },
                 },
-            }),
-        ))
-        session.add(MafRoundtableRunRow(
-            id="v2-run", session_id="v2-sess", status="pending",
-            created_at=now, updated_at=now,
-        ))
-        # Persist ALL candidates (simulating old V2 behavior).
-        from deerflow.persistence.maf_roundtable.model import MafRoundtableParticipantRow
-        for i, (agent_id, role) in enumerate([(MANAGER, "manager"), ("seat-a", "seat"), ("seat-b", "seat"), ("seat-c", "seat")]):
-            session.add(MafRoundtableParticipantRow(
-                id=f"p-{i}", session_id="v2-sess", agent_id=agent_id,
-                role=role, sort_order=i, created_at=now, updated_at=now,
             ))
-        await session.commit()
+            session.add(MafRoundtableRunRow(
-    yield service
+                id="v2-run", session_id="v2-sess", status="pending",
-    await engine.dispose()
+                created_at=now, updated_at=now,
+            ))
+            # Persist ALL candidates (simulating old V2 behavior).
+            for i, (agent_id, role) in enumerate([(MANAGER, "manager"), ("seat-a", "seat"), ("seat-b", "seat"), ("seat-c", "seat")]):
+                session.add(MafRoundtableParticipantRow(
+                    id=f"p-{i}", session_id="v2-sess", agent_id=agent_id,
+                    role=role, sort_order=i, created_at=now, updated_at=now,
+                ))
+            await session.commit()
+        yield service
+    finally:
+        await engine.dispose()
 
 
-def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict):
+def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict | None = None):
-    """Create a WorkflowRunner with a mocked adapter for V2 tests."""
+    """Create a WorkflowRunner with a mocked adapter for V2 tests.
+
+    When ``dispatch_decision`` is None any model call raises — used by
+    recovery tests that must prove the manager is never re-invoked.
+    """
     from types import SimpleNamespace
-    from unittest.mock import AsyncMock
 
     from app.maf_roundtable.workflow_runner import WorkflowRunner
 
@@ -324,13 +333,15 @@ def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict):
 
     async def _run_seat(**kwargs):
         calls.append(kwargs)
+        if dispatch_decision is None:
+            raise AssertionError("manager/seat model must not be called in this test")
         stage = kwargs.get("execution_stage", "")
         agent_id = kwargs["agent_id"]
         if stage == "manager_dispatch":
             return SeatResponse(agent_id=agent_id, turn_index=0, content=json.dumps(dispatch_decision), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
         if stage == "direct_answer":
             return SeatResponse(agent_id=agent_id, turn_index=0, content="manager direct answer", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
-        if stage == "expert":
+        if stage in ("expert", "group_chat"):
             return SeatResponse(agent_id=agent_id, turn_index=1, content=json.dumps({"summary": f"{agent_id} result", "findings": ["f"]}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
         if stage == "cross_review":
             return SeatResponse(agent_id=agent_id, turn_index=2, content=json.dumps({"summary": "ok", "confirmed_candidates": []}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
@@ -343,7 +354,18 @@ def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict):
     async def _ensure_threads(session_id, run_id, participants):
         agent_ids = [p["agent_id"] for p in participants]
         ensure_threads_calls.append(agent_ids)
-        return {p["agent_id"]: f"thread-{p['agent_id']}" for p in participants}
+
+        result = {}
+        for participant in participants:
+            agent_id = participant["agent_id"]
+            thread_id = f"thread-{agent_id}"
+            await v2_svc.store.participants.bind_thread(
+                session_id,
+                agent_id,
+                thread_id=thread_id,
+            )
+            result[agent_id] = thread_id
+        return result
 
     monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat))
     monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(side_effect=_ensure_threads))
@@ -395,11 +417,23 @@ async def test_runner_parallel_consultation_only_selected_seats(v2_svc, monkeypa
     final_run = await v2_svc.store.runs.get_run("v2-run")
     assert final_run["status"] == "completed"
 
+    # 5. Database state: threads are bound ONLY for manager + selected seats.
+    # seat-c keeps its historical participant row (P1-3 never deletes history),
+    # but it must have zero seat threads and zero seat execution.
+    participants = await v2_svc.store.participants.list_participants("v2-sess")
+    by_agent = {row["agent_id"]: row for row in participants}
+
+    assert by_agent[MANAGER]["thread_id"] == f"thread-{MANAGER}"
+    assert by_agent["seat-a"]["thread_id"] == "thread-seat-a"
+    assert by_agent["seat-b"]["thread_id"] == "thread-seat-b"
+    assert by_agent["seat-c"]["thread_id"] is None
+
 
 @pytest.mark.asyncio
 async def test_runner_direct_answer_zero_seats(v2_svc, monkeypatch):
-    """direct_answer: manager_only — zero seat participants, zero seat threads,
+    """direct_answer: manager_only — zero seat threads, zero seat execution;
-    zero seat runs; final answer comes from manager."""
+    historical seat participant rows may remain, but never get threads; final
+    answer comes from manager."""
     pytest.importorskip("agent_framework")
 
     dispatch_decision = {
@@ -482,6 +516,158 @@ async def test_runner_clarification_waiting_input(v2_svc, monkeypatch):
     assert plan.get("clarification_prompt") == "请明确分析的目标市场和时间范围。"
 
 
+@pytest.mark.asyncio
+async def test_runner_group_chat_executes_only_selected_seats(v2_svc, monkeypatch):
+    """group_chat: history participants are manager+A+B+C; the manager selects
+    only A/B with execution_mode=group_chat. Executed through
+    WorkflowRunner.execute(): initial ensure_threads gets only the manager,
+    only A/B later receive threads and speak as coordinator speakers, and C
+    never gets a thread or a group_chat execution stage."""
+    pytest.importorskip("agent_framework")
+
+    # Switch the fixture session to a group_chat route.
+    await v2_svc.store.sessions.update_session("v2-sess", user_id="u1", fields={
+        "mode": "group_chat",
+        "max_rounds": 2,
+        "task_payload": {
+            "maf_v2": {
+                "dispatch_contract_enabled": True,
+                "candidate_set": {
+                    "manager_agent_id": MANAGER,
+                    "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"],
+                    "mode": "group_chat",
+                    "max_rounds": 2,
+                },
+            },
+        },
+    })
+
+    dispatch_decision = {
+        "intent_type": "parallel_consultation",
+        "selected_seat_agent_ids": ["seat-a", "seat-b"],
+        "assignments": [
+            {"agent_id": "seat-a", "task_goal": "market analysis", "delivery_requirements": ["data"], "order": 1},
+            {"agent_id": "seat-b", "task_goal": "competitor review", "delivery_requirements": ["sources"], "order": 2},
+        ],
+        "execution_mode": "group_chat",
+        "convergence_path": ["manager synthesis"],
+        "reason": "group discussion needed",
+    }
+
+    runner, calls, ensure_threads_calls = _make_v2_runner(v2_svc, monkeypatch, dispatch_decision=dispatch_decision)
+
+    session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
+    run = await v2_svc.store.runs.get_run("v2-run")
+    await runner.execute(session, run)
+
+    # 1. First ensure_threads call: only manager.
+    assert ensure_threads_calls[0] == [MANAGER], \
+        f"First ensure_threads must only get manager, got {ensure_threads_calls[0]}"
+
+    # 2. seat-c never appears in any ensure_threads call.
+    all_ensured = [aid for call in ensure_threads_calls for aid in call]
+    assert "seat-c" not in all_ensured, "seat-c must never receive a thread"
+
+    # 3. Only A/B speak in the group_chat stage; C never executes.
+    speakers = [c["agent_id"] for c in calls if c.get("execution_stage") == "group_chat"]
+    assert speakers, "selected group_chat seats must speak"
+    assert set(speakers) == {"seat-a", "seat-b"}, f"unexpected speakers {speakers}"
+    assert "seat-c" not in speakers
+
+    # 4. Run completed.
+    final_run = await v2_svc.store.runs.get_run("v2-run")
+    assert final_run["status"] == "completed"
+
+    # 5. Database state: threads bound only for manager + selected seats; the
+    # historical seat-c participant row remains but holds zero seat threads.
+    participants = await v2_svc.store.participants.list_participants("v2-sess")
+    by_agent = {row["agent_id"]: row for row in participants}
+    assert by_agent[MANAGER]["thread_id"] == f"thread-{MANAGER}"
+    assert by_agent["seat-a"]["thread_id"] == "thread-seat-a"
+    assert by_agent["seat-b"]["thread_id"] == "thread-seat-b"
+    assert by_agent["seat-c"]["thread_id"] is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("has_prompt", [True, False], ids=["with_prompt", "without_prompt"])
+async def test_runner_recover_clarification_checkpoint(v2_svc, monkeypatch, has_prompt):
+    """WorkflowRunner.recover() on a run stuck 'running' whose checkpoint is a
+    completed clarification dispatch must land run+session in waiting_input,
+    never re-call the manager model, and never create seat threads. A valid
+    original prompt survives untouched; a missing prompt degrades to the
+    unified safe clarification."""
+    pytest.importorskip("agent_framework")
+
+    original_prompt = "请明确分析的目标市场和时间范围。"
+    dispatch_plan = {
+        "execution_mode": "manager_only",
+        "convergence_path": ["clarification"],
+        "reason": "insufficient_task_detail",
+        "source": "manager",
+    }
+    if has_prompt:
+        dispatch_plan["clarification_prompt"] = original_prompt
+
+    state = RoundtableState(
+        session_id="v2-sess", run_id="v2-run", task_text="分析市场趋势",
+        mode="structured", model_name=None, reasoning_effort=None,
+        max_rounds=8, write_policy="auto", manager_agent_id=MANAGER,
+        seat_agent_ids=["seat-a", "seat-b", "seat-c"],
+        thread_map={}, capability_snapshot=None,
+        seat_timeout_seconds=5, schema_repair_enabled=True,
+        intent_type="clarification",
+        dispatch_plan=dispatch_plan,
+        dispatch_contract_enabled=True,
+    )
+    state.selected_seat_agent_ids = []
+    state.status = "waiting_input"
+    state.stage = "waiting_input"
+    state.mark_completed("manager_dispatch")
+
+    # Simulate a crash: the run row is stuck "running" while the checkpoint
+    # already records a completed clarification dispatch.
+    await v2_svc.store.runs.update_run("v2-run", {
+        "status": "running",
+        "maf_checkpoint": state.to_dict(),
+    })
+
+    # dispatch_decision=None → any model call raises.
+    runner, calls, ensure_threads_calls = _make_v2_runner(v2_svc, monkeypatch)
+
+    final_run = await runner.recover("v2-run")
+
+    # 1. Run and session both end as waiting_input.
+    assert final_run["status"] == "waiting_input"
+    final_session = await v2_svc.store.sessions.get_session("v2-sess", user_id="u1")
+    assert final_session["status"] == "waiting_input"
+
+    # 2. Manager model was never re-invoked.
+    assert calls == []
+    runner._adapter.run_seat_turn.assert_not_awaited()
+
+    # 3. No seat threads — only the manager may hold a thread.
+    ensured = [aid for call in ensure_threads_calls for aid in call]
+    assert "seat-a" not in ensured
+    assert "seat-b" not in ensured
+    assert "seat-c" not in ensured
+    participants = await v2_svc.store.participants.list_participants("v2-sess")
+    by_agent = {row["agent_id"]: row for row in participants}
+    for seat_id in ("seat-a", "seat-b", "seat-c"):
+        assert by_agent[seat_id]["thread_id"] is None
+
+    # 4. Checkpoint outcome: valid prompt preserved / missing prompt degraded.
+    checkpoint = final_run.get("maf_checkpoint") or {}
+    assert checkpoint.get("intent_type") == "clarification"
+    assert checkpoint.get("status") == "waiting_input"
+    plan = checkpoint.get("dispatch_plan") or {}
+    if has_prompt:
+        assert plan.get("clarification_prompt") == original_prompt
+        assert plan.get("reason") == "insufficient_task_detail"
+    else:
+        assert plan.get("clarification_prompt")
+        assert plan.get("reason") == "checkpoint_clarification_prompt_missing"
+
+
 # ── P1-3: _participants_requiring_threads unit tests ────────────────────────
 
 
@@ -513,8 +699,6 @@ def test_participants_requiring_threads_legacy_returns_all():
 
 def test_participants_requiring_threads_v2_no_plan_only_manager():
     """V2 session without a validated plan: only manager gets threads."""
-    from types import SimpleNamespace
-
     from app.maf_roundtable.workflow_runner import WorkflowRunner
 
     runner = WorkflowRunner.__new__(WorkflowRunner)