董宏凯
1aeee86984
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema. P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events. P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback). P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks. P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
779 lines
33 KiB
Python
779 lines
33 KiB
Python
"""Phase-3 Round-2 acceptance tests (P0-3): real manager workflow coexistence.
|
||
|
||
Drives the REAL workflow executors (PlannerExecutor → SeatExecutor ×N →
|
||
CrossReviewExecutor → ApprovalGateExecutor → FinalizerExecutor) through a
|
||
manager + 2-seat structured run, with a fake DeerFlow adapter whose
|
||
``run_seat_turn`` records each call's ``execution_stage`` into a REAL seat-run
|
||
store. This proves the Phase-3 disambiguation invariants end-to-end:
|
||
|
||
- manager planner / cross_review / finalizer all share the SAME business round
|
||
(turn_index) but occupy distinct seat-run rows via ``execution_stage`` — they
|
||
must coexist without a four-tuple collision.
|
||
- an expert seat turn and its one-shot ``format_repair`` retry coexist for the
|
||
same (agent, turn).
|
||
- invalid expert output triggers exactly one ``format_repair`` DeerFlow call,
|
||
and both rows persist.
|
||
- invalid cross-review output triggers exactly one ``cross_review_repair``
|
||
DeerFlow call, and ``cross_review`` + ``cross_review_repair`` rows coexist.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import datetime
|
||
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.deerflow_agent_adapter import SeatResponse
|
||
from app.maf_roundtable.workflow_executors import (
|
||
ApprovalGateExecutor,
|
||
CrossReviewExecutor,
|
||
FinalizerExecutor,
|
||
PlannerExecutor,
|
||
RoundtableState,
|
||
SeatExecutor,
|
||
)
|
||
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
||
from deerflow.persistence.maf_roundtable.model import (
|
||
Base,
|
||
MafRoundtableParticipantRow,
|
||
MafRoundtableRunRow,
|
||
MafRoundtableSessionRow,
|
||
)
|
||
|
||
MANAGER = "manager-1"
|
||
SEATS = ["seat-a", "seat-b"]
|
||
|
||
|
||
class _FakeCtx:
|
||
def __init__(self) -> None:
|
||
self.messages: list[object] = []
|
||
self.outputs: list[object] = []
|
||
|
||
async def send_message(self, msg: object) -> None:
|
||
self.messages.append(msg)
|
||
|
||
async def yield_output(self, msg: object) -> None:
|
||
self.outputs.append(msg)
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def svc():
|
||
engine = create_async_engine("sqlite+aiosqlite://", 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="sess-1", user_id="u1", title="t", task_text="分析市场",
|
||
mode="structured", status="running", write_policy="auto",
|
||
manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
|
||
))
|
||
session.add(MafRoundtableRunRow(
|
||
id="run-1", session_id="sess-1", status="running",
|
||
created_at=now, updated_at=now,
|
||
))
|
||
await session.commit()
|
||
yield service
|
||
await engine.dispose()
|
||
|
||
|
||
def _state() -> RoundtableState:
|
||
return RoundtableState(
|
||
session_id="sess-1", run_id="run-1", task_text="分析市场",
|
||
mode="structured", model_name=None, reasoning_effort=None,
|
||
max_rounds=8, write_policy="auto", manager_agent_id=MANAGER,
|
||
seat_agent_ids=SEATS,
|
||
thread_map={MANAGER: "t-mgr", "seat-a": "t-a", "seat-b": "t-b"},
|
||
capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True,
|
||
)
|
||
|
||
|
||
def _resp(content: str) -> SeatResponse:
|
||
return SeatResponse(
|
||
agent_id="x", turn_index=0, content=content,
|
||
tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1},
|
||
)
|
||
|
||
|
||
class _RecordingAdapter:
|
||
"""Fake adapter that (a) records every run_seat_turn's execution_stage and
|
||
(b) writes a REAL seat-run row for each call so the four-tuple coexistence
|
||
invariant is checked against the actual DB."""
|
||
|
||
def __init__(self, service) -> None:
|
||
self._service = service
|
||
self.calls: list[tuple[str, str, int, str]] = [] # (agent, turn, stage)
|
||
self._seat_outputs: dict[tuple[str, str], str] = {} # (agent, stage) -> content
|
||
self._user_id = "u1"
|
||
self._emit_critical = AsyncMock()
|
||
self._emit = AsyncMock()
|
||
|
||
def set_output(self, agent_id: str, execution_stage: str, content: str) -> None:
|
||
self._seat_outputs[(agent_id, execution_stage)] = content
|
||
|
||
async def run_seat_turn(self, **kwargs):
|
||
agent_id = kwargs["agent_id"]
|
||
turn_index = kwargs["turn_index"]
|
||
stage = kwargs.get("execution_stage", "expert")
|
||
self.calls.append((agent_id, turn_index, stage))
|
||
# Persist a seat-run row so coexistence is verified against the real DB.
|
||
await self._service.store.seat_runs.create_seat_run(
|
||
maf_run_id=kwargs["run_id"], agent_id=agent_id,
|
||
turn_index=turn_index, execution_stage=stage,
|
||
)
|
||
content = self._seat_outputs.get((agent_id, stage), '{"summary":"ok"}')
|
||
return _resp(content)
|
||
|
||
|
||
# ── 1. Full structured run: manager stages coexist ───────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_structured_run_manager_stages_coexist(svc):
|
||
"""A complete planner→experts→cross_review→approval→finalizer run must
|
||
persist manager planner/cross_review/finalizer rows that ALL share the same
|
||
business round (turn_index) but differ by execution_stage — no collision."""
|
||
adapter = _RecordingAdapter(svc)
|
||
# Manager emits valid JSON at each stage.
|
||
adapter.set_output(MANAGER, "planner", "分工：seat-a 负责需求，seat-b 负责竞品")
|
||
adapter.set_output(MANAGER, "cross_review", json.dumps({
|
||
"summary": "审查通过",
|
||
"evidence_ok": True,
|
||
"confirmed_candidates": [],
|
||
}))
|
||
adapter.set_output(MANAGER, "finalizer", "最终结论：市场看好")
|
||
# Seats emit valid structured output.
|
||
for s in SEATS:
|
||
adapter.set_output(s, "expert", json.dumps({"summary": f"{s} 结论", "findings": ["f"]}))
|
||
|
||
state = _state()
|
||
ctx = _FakeCtx()
|
||
await PlannerExecutor(adapter, state).handle("task", ctx)
|
||
for s in SEATS:
|
||
await SeatExecutor(adapter, state, s).handle({}, _FakeCtx())
|
||
await CrossReviewExecutor(adapter, state).handle({}, _FakeCtx())
|
||
await ApprovalGateExecutor(adapter, state, svc).handle({}, _FakeCtx())
|
||
await FinalizerExecutor(adapter, state).handle({}, _FakeCtx())
|
||
|
||
# Assert every manager stage produced a distinct seat-run row, all sharing
|
||
# turn_index (planner=0, cross_review/finalizer advance current_round; the
|
||
# invariant is that the SAME (agent) has multiple distinct stages).
|
||
rows = await svc.store.seat_runs.get_by_maf_run("run-1")
|
||
mgr_rows = [r for r in rows if r["agent_id"] == MANAGER]
|
||
mgr_stages = {r["execution_stage"] for r in mgr_rows}
|
||
assert {"planner", "cross_review", "finalizer"}.issubset(mgr_stages), \
|
||
f"manager stages missing; got {mgr_stages}"
|
||
# No duplicate four-tuple (the unique constraint would have raised).
|
||
seen = set()
|
||
for r in mgr_rows:
|
||
key = (r["maf_run_id"], r["agent_id"], r["turn_index"], r["execution_stage"])
|
||
assert key not in seen, f"duplicate four-tuple {key}"
|
||
seen.add(key)
|
||
# The final result was produced.
|
||
assert state.final_result is not None
|
||
assert state.status == "completed"
|
||
|
||
|
||
# ── 2. expert + format_repair coexist ────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_expert_format_repair_coexist(svc):
|
||
"""A seat that emits invalid JSON triggers exactly one ``format_repair``
|
||
DeerFlow call; both the expert and the format_repair rows persist for the
|
||
same (agent, turn) without colliding."""
|
||
adapter = _RecordingAdapter(svc)
|
||
# First expert call: invalid JSON → triggers repair. The format_repair call
|
||
# (stage='format_repair') returns valid JSON.
|
||
adapter.set_output("seat-a", "expert", "not-json-at-all")
|
||
adapter.set_output("seat-a", "format_repair", json.dumps({"summary": "repaired"}))
|
||
|
||
state = _state()
|
||
state.current_round = 1
|
||
state.plan = {"assignments": {"seat-a": {"instruction": "x", "role": "A"}}}
|
||
|
||
await SeatExecutor(adapter, state, "seat-a").handle({}, _FakeCtx())
|
||
|
||
stage_calls = [(a, t, s) for (a, t, s) in adapter.calls if a == "seat-a"]
|
||
stages = [s for _, _, s in stage_calls]
|
||
assert "expert" in stages and "format_repair" in stages
|
||
# Exactly one repair (not a retry loop).
|
||
assert stages.count("format_repair") == 1
|
||
|
||
rows = await svc.store.seat_runs.get_by_maf_run("run-1")
|
||
seat_a_stages = {r["execution_stage"] for r in rows if r["agent_id"] == "seat-a"}
|
||
assert seat_a_stages == {"expert", "format_repair"}
|
||
|
||
|
||
# ── 3. cross_review + cross_review_repair coexist ────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cross_review_repair_coexist(svc):
|
||
"""An invalid cross-review output triggers exactly one ``cross_review_repair``
|
||
DeerFlow call; cross_review and cross_review_repair rows coexist for the
|
||
manager."""
|
||
adapter = _RecordingAdapter(svc)
|
||
adapter.set_output(MANAGER, "cross_review", "totally not json")
|
||
adapter.set_output(MANAGER, "cross_review_repair", json.dumps({
|
||
"summary": "repaired review", "confirmed_candidates": [],
|
||
}))
|
||
|
||
state = _state()
|
||
# Pretend experts already ran so the cross-review barrier sees outcomes.
|
||
state.seat_outcomes = {s: {"status": "completed"} for s in SEATS}
|
||
state.current_round = 1
|
||
|
||
await CrossReviewExecutor(adapter, state).handle({}, _FakeCtx())
|
||
|
||
mgr_stages = [s for a, _, s in adapter.calls if a == MANAGER]
|
||
assert "cross_review" in mgr_stages and "cross_review_repair" in mgr_stages
|
||
assert mgr_stages.count("cross_review_repair") == 1
|
||
|
||
rows = await svc.store.seat_runs.get_by_maf_run("run-1")
|
||
mgr_row_stages = {r["execution_stage"] for r in rows if r["agent_id"] == MANAGER}
|
||
assert {"cross_review", "cross_review_repair"}.issubset(mgr_row_stages)
|
||
|
||
# The repaired review was applied (deterministic degradation did NOT happen).
|
||
assert state.cross_review_result is not None
|
||
assert state.cross_review_result.get("schema_status") == "repaired"
|
||
assert state.cross_review_result.get("summary") == "repaired review"
|
||
|
||
|
||
# ── 4. manager present in participants-style config ──────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_manager_role_wired_through_state(svc):
|
||
"""Sanity: RoundtableState carries manager_agent_id and the planner uses it
|
||
for the planner seat-turn (not a seat id)."""
|
||
state = _state()
|
||
assert state.manager_agent_id == MANAGER
|
||
assert MANAGER not in state.seat_agent_ids
|
||
|
||
adapter = _RecordingAdapter(svc)
|
||
adapter.set_output(MANAGER, "planner", "plan")
|
||
ctx = _FakeCtx()
|
||
await PlannerExecutor(adapter, state).handle("task", ctx)
|
||
|
||
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)
|
||
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,
|
||
# Stage-3 topology tests freeze the pre-Stage-4 behavior:
|
||
# plain seat fan-out, no quality gate.
|
||
"quality_gate_enabled": False,
|
||
"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).
|
||
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 | 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 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)
|
||
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 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})
|
||
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)
|
||
|
||
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))
|
||
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"
|
||
|
||
# 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 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 = {
|
||
"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") == "请明确分析的目标市场和时间范围。"
|
||
|
||
|
||
@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 ────────────────────────
|
||
|
||
|
||
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 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
|