diff --git a/offline-backend-20260512/backend/tests/test_maf_phase3_manager_workflow.py b/offline-backend-20260512/backend/tests/test_maf_phase3_manager_workflow.py
index 975ffba7..7ca25938 100644
--- a/offline-backend-20260512/backend/tests/test_maf_phase3_manager_workflow.py
+++ b/offline-backend-20260512/backend/tests/test_maf_phase3_manager_workflow.py
@@ -262,3 +262,330 @@ async def test_manager_role_wired_through_state(svc):
     planner_calls = [(a, s) for a, _, s in adapter.calls if s == "planner"]
     assert planner_calls == [(MANAGER, "planner")], \
         "planner stage must run as the MANAGER, not a seat"
+
+
+# ── P1-3 / P1-4: Runner-level full chain tests ──────────────────────────────
+
+
+@pytest_asyncio.fixture
+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:
+        await conn.run_sync(Base.metadata.create_all)
+    sf = async_sessionmaker(engine, expire_on_commit=False)
+    store = make_maf_roundtable_store(sf)
+    service = make_maf_service(store)
+    now = datetime(2026, 1, 1)
+    async with sf() as session:
+        session.add(MafRoundtableSessionRow(
+            id="v2-sess", user_id="u1", title="v2", task_text="分析市场趋势",
+            mode="structured", status="pending", write_policy="auto",
+            manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
+            task_payload=json.dumps({
+                "maf_v2": {
+                    "dispatch_contract_enabled": True,
+                    "candidate_set": {
+                        "manager_agent_id": MANAGER,
+                        "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"],
+                        "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()
+    yield service
+    await engine.dispose()
+
+
+def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict):
+    """Create a WorkflowRunner with a mocked adapter for V2 tests."""
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    from app.maf_roundtable.workflow_runner import WorkflowRunner
+
+    app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=v2_svc, checkpointer=None))
+    runner = WorkflowRunner(app, service=v2_svc, user_id="u1")
+
+    calls: list[dict] = []
+
+    async def _run_seat(**kwargs):
+        calls.append(kwargs)
+        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":
+            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})
+        if stage == "finalizer":
+            return SeatResponse(agent_id=agent_id, turn_index=3, content="final report", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
+        return SeatResponse(agent_id=agent_id, turn_index=0, content='{"summary":"ok"}', tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
+
+    ensure_threads_calls: list[list[str]] = []
+
+    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}
+
+    monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat))
+    monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(side_effect=_ensure_threads))
+    monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock())
+    monkeypatch.setattr(runner._adapter, "_emit", AsyncMock())
+    return runner, calls, ensure_threads_calls
+
+
+@pytest.mark.asyncio
+async def test_runner_parallel_consultation_only_selected_seats(v2_svc, monkeypatch):
+    """parallel_consultation: initial ensure_threads only gets manager;
+    after dispatch, only selected seats A/B get threads; C never appears."""
+    pytest.importorskip("agent_framework")
+
+    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": "parallel",
+        "convergence_path": ["manager synthesis"],
+        "reason": "two experts 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 len(ensure_threads_calls) >= 1
+    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. seat-a and seat-b were executed (expert stage).
+    expert_agents = [c["agent_id"] for c in calls if c.get("execution_stage") == "expert"]
+    assert "seat-a" in expert_agents
+    assert "seat-b" in expert_agents
+    assert "seat-c" not in expert_agents
+
+    # 4. Run completed.
+    final_run = await v2_svc.store.runs.get_run("v2-run")
+    assert final_run["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_runner_direct_answer_zero_seats(v2_svc, monkeypatch):
+    """direct_answer: manager_only — zero seat participants, zero seat threads,
+    zero seat runs; final answer comes from manager."""
+    pytest.importorskip("agent_framework")
+
+    dispatch_decision = {
+        "intent_type": "direct_answer",
+        "selected_seat_agent_ids": [],
+        "assignments": [],
+        "execution_mode": "manager_only",
+        "convergence_path": ["manager answer"],
+        "reason": "simple question",
+    }
+
+    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. Only manager gets a thread.
+    all_ensured = [aid for call in ensure_threads_calls for aid in call]
+    assert all_ensured == [MANAGER], f"Only manager should get a thread, got {all_ensured}"
+
+    # 2. No seat was executed (no expert stage calls).
+    expert_calls = [c for c in calls if c.get("execution_stage") == "expert"]
+    assert expert_calls == [], "No seat should execute in direct_answer mode"
+
+    # 3. Manager produced the answer via direct_answer stage.
+    direct_calls = [c for c in calls if c.get("execution_stage") == "direct_answer"]
+    assert len(direct_calls) == 1
+    assert direct_calls[0]["agent_id"] == MANAGER
+
+    # 4. Run completed.
+    final_run = await v2_svc.store.runs.get_run("v2-run")
+    assert final_run["status"] == "completed"
+    # 5. Zero seat runs in DB.
+    seat_runs = await v2_svc.store.seat_runs.get_by_maf_run("v2-run")
+    expert_rows = [r for r in seat_runs if r["execution_stage"] == "expert"]
+    assert expert_rows == []
+
+
+@pytest.mark.asyncio
+async def test_runner_clarification_waiting_input(v2_svc, monkeypatch):
+    """clarification: manager returns clarification — no seats started,
+    run/session end as waiting_input."""
+    pytest.importorskip("agent_framework")
+
+    dispatch_decision = {
+        "intent_type": "clarification",
+        "selected_seat_agent_ids": [],
+        "assignments": [],
+        "execution_mode": "manager_only",
+        "convergence_path": ["clarification"],
+        "reason": "task unclear",
+        "clarification_prompt": "请明确分析的目标市场和时间范围。",
+    }
+
+    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. Only manager gets a thread.
+    all_ensured = [aid for call in ensure_threads_calls for aid in call]
+    assert all_ensured == [MANAGER]
+
+    # 2. No seat executed.
+    expert_calls = [c for c in calls if c.get("execution_stage") == "expert"]
+    assert expert_calls == []
+
+    # 3. Run and session are waiting_input.
+    final_run = await v2_svc.store.runs.get_run("v2-run")
+    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"
+
+    # 4. Checkpoint preserves the clarification prompt.
+    checkpoint = final_run.get("maf_checkpoint") or {}
+    assert checkpoint.get("intent_type") == "clarification"
+    plan = checkpoint.get("dispatch_plan") or {}
+    assert plan.get("clarification_prompt") == "请明确分析的目标市场和时间范围。"
+
+
+# ── P1-3: _participants_requiring_threads unit tests ────────────────────────
+
+
+def test_participants_requiring_threads_legacy_returns_all():
+    """Legacy MAF sessions without V2 contract keep original behavior."""
+    from types import SimpleNamespace
+
+    from app.maf_roundtable.workflow_runner import WorkflowRunner
+
+    app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=None, checkpointer=None))
+    runner = WorkflowRunner.__new__(WorkflowRunner)
+    runner._app = app
+    runner._service = None
+    runner._user_id = "u1"
+
+    participants = [
+        {"agent_id": "manager", "role": "manager"},
+        {"agent_id": "seat-a", "role": "seat"},
+        {"agent_id": "seat-b", "role": "seat"},
+    ]
+    result = runner._participants_requiring_threads(
+        participants,
+        dispatch_contract_enabled=False,
+        dispatch_plan=None,
+        intent_type=None,
+    )
+    assert result == participants
+
+
+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)
+    runner._app = None
+    runner._service = None
+    runner._user_id = "u1"
+
+    participants = [
+        {"agent_id": "manager", "role": "manager"},
+        {"agent_id": "seat-a", "role": "seat"},
+        {"agent_id": "seat-b", "role": "seat"},
+        {"agent_id": "seat-c", "role": "seat"},
+    ]
+    result = runner._participants_requiring_threads(
+        participants,
+        dispatch_contract_enabled=True,
+        dispatch_plan=None,
+        intent_type=None,
+    )
+    assert [p["agent_id"] for p in result] == ["manager"]
+
+
+def test_participants_requiring_threads_v2_direct_answer_only_manager():
+    """V2 with validated direct_answer plan: only manager."""
+    from app.maf_roundtable.workflow_runner import WorkflowRunner
+
+    runner = WorkflowRunner.__new__(WorkflowRunner)
+    runner._app = None
+    runner._service = None
+    runner._user_id = "u1"
+
+    participants = [
+        {"agent_id": "manager", "role": "manager"},
+        {"agent_id": "seat-a", "role": "seat"},
+    ]
+    result = runner._participants_requiring_threads(
+        participants,
+        dispatch_contract_enabled=True,
+        dispatch_plan={"intent_type": "direct_answer", "execution_mode": "manager_only"},
+        intent_type="direct_answer",
+    )
+    assert [p["agent_id"] for p in result] == ["manager"]
+
+
+def test_participants_requiring_threads_v2_selected_seats():
+    """V2 with validated parallel plan: manager + selected seats only."""
+    from app.maf_roundtable.workflow_runner import WorkflowRunner
+
+    runner = WorkflowRunner.__new__(WorkflowRunner)
+    runner._app = None
+    runner._service = None
+    runner._user_id = "u1"
+
+    participants = [
+        {"agent_id": "manager", "role": "manager"},
+        {"agent_id": "seat-a", "role": "seat"},
+        {"agent_id": "seat-b", "role": "seat"},
+        {"agent_id": "seat-c", "role": "seat"},
+    ]
+    result = runner._participants_requiring_threads(
+        participants,
+        dispatch_contract_enabled=True,
+        dispatch_plan={
+            "intent_type": "parallel_consultation",
+            "execution_mode": "parallel",
+            "seat_agent_ids": ["seat-a", "seat-b"],
+        },
+        intent_type="parallel_consultation",
+    )
+    agent_ids = [p["agent_id"] for p in result]
+    assert "manager" in agent_ids
+    assert "seat-a" in agent_ids
+    assert "seat-b" in agent_ids
+    assert "seat-c" not in agent_ids