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 a88d896e..ee319ab7 100644 --- a/offline-backend-20260512/backend/app/maf_roundtable/workflow_executors.py +++ b/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(), 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 c1e02182..c93f5c41 100644 --- a/offline-backend-20260512/backend/app/maf_roundtable/workflow_runner.py +++ b/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), 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 cfafb02a..799e63a4 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 @@ -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 = [ { 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 index 63503e0f..1444a26d 100644 --- a/offline-backend-20260512/backend/tests/test_maf_phase3_dispatch_recovery.py +++ b/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 ───────────────────────────────────────────────────────────────── 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 7ca25938..431f4c26 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 @@ -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: - 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, + try: + 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={ + "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() + 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). + 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): - """Create a WorkflowRunner with a mocked adapter for V2 tests.""" +def _make_v2_runner(v2_svc, monkeypatch, *, dispatch_decision: dict | None = None): + """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, - zero seat runs; final answer comes from manager.""" + """direct_answer: manager_only — zero seat threads, zero seat execution; + 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)