"""Phase 3 structured workflow acceptance tests (MAF-03).""" from __future__ import annotations import json from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock import pytest from maf_test_helpers import make_maf_service from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.maf_roundtable.fact_ledger import executed_action_ids, proposed_action_ids from app.maf_roundtable.workflow_executors import ( ApprovalGateExecutor, CoordinatorExecutor, CrossReviewExecutor, DispatchPreparationTerminalExecutor, FinalizerExecutor, ManagerAnswerExecutor, ManagerDispatchExecutor, PlannerExecutor, RoundtableState, SeatExecutor, WaitingTerminalExecutor, ) from app.maf_roundtable.workflow_factory import build_manager_dispatch_workflow, create_workflow from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store class _FakeCtx: def __init__(self) -> None: self.messages: list[Any] = [] self.outputs: list[Any] = [] async def send_message(self, msg: Any) -> None: self.messages.append(msg) async def yield_output(self, msg: Any) -> None: self.outputs.append(msg) class _FakeSeatResponse: def __init__(self, content: str = "", tool_calls: list | None = None, error: str | None = None) -> None: self.content = content self.tool_calls = tool_calls or [] self.error = error self.usage = {"input_tokens": 1, "output_tokens": 1} def _state(seat_ids: list[str] | None = None) -> RoundtableState: seats = seat_ids or ["seat-a", "seat-b", "seat-c"] return 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=seats, thread_map={a: f"t-{a}" for a in seats}, capability_snapshot=None, seat_timeout_seconds=5, schema_repair_enabled=True, ) @pytest.mark.asyncio async def test_direct_answer_is_manager_only(): state = _state([]) state.manager_agent_id = "manager" state.thread_map = {"manager": "manager-thread"} adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content="direct answer")), _emit_critical=AsyncMock(), ) await ManagerAnswerExecutor(adapter, state).handle("task", _FakeCtx()) adapter.run_seat_turn.assert_awaited_once() assert adapter.run_seat_turn.await_args.kwargs["agent_id"] == "manager" assert adapter.run_seat_turn.await_args.kwargs["execution_stage"] == "direct_answer" assert state.status == "completed" dispatch = next( call.args[1] for call in adapter._emit_critical.await_args_list if call.args[1].get("event_type") == "manager.dispatch" ) assert dispatch["payload"]["execution_mode"] == "manager_only" assert dispatch["payload"]["selected_seat_agent_ids"] == [] @pytest.mark.asyncio async def test_manager_only_topology_runs_real_maf_workflow_without_seats(): pytest.importorskip("agent_framework") state = _state([]) state.manager_agent_id = "manager" state.thread_map = {"manager": "manager-thread"} state.intent_type = "direct_answer" state.dispatch_plan = { "manager_agent_id": "manager", "seat_agent_ids": [], "assignments": [], "intent_type": "direct_answer", "max_rounds": 8, "candidate_seat_agent_ids": ["seat-a", "seat-b", "seat-c"], "execution_mode": "manager_only", "convergence_path": ["manager answer"], } adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content="manager answer")), _emit_critical=AsyncMock(), ) manager = ManagerAnswerExecutor(adapter, state) workflow = create_workflow( mode="manager_only", manager_answer=manager, seat_executors=[object(), object()], approval_gate=object(), finalizer=object(), waiting_terminal=object(), seat_agent_ids=["seat-a", "seat-b"], ) async for _ in workflow.run(state.task_text, stream=True, include_status_events=True): pass adapter.run_seat_turn.assert_awaited_once() assert adapter.run_seat_turn.await_args.kwargs["agent_id"] == "manager" assert state.status == "completed" @pytest.mark.asyncio async def test_v2_direct_answer_runs_manager_dispatch_then_manager_only_workflow(): pytest.importorskip("agent_framework") state = _state(["seat-a", "seat-b"]) state.thread_map = {"manager": "manager-thread"} state.manager_agent_id = "manager" state.candidate_seat_agent_ids = ["seat-a", "seat-b"] state.dispatch_contract_enabled = True decision = { "intent_type": "direct_answer", "selected_seat_agent_ids": [], "assignments": [], "execution_mode": "manager_only", "convergence_path": ["manager answer"], "reason": "manager can answer directly", } async def run_turn(**kwargs): if kwargs["execution_stage"] == "manager_dispatch": return _FakeSeatResponse(content=json.dumps(decision)) return _FakeSeatResponse(content="direct answer") adapter = SimpleNamespace( run_seat_turn=AsyncMock(side_effect=run_turn), _emit_critical=AsyncMock(), ) preparation = build_manager_dispatch_workflow( manager_dispatch=ManagerDispatchExecutor(adapter, state), terminal=DispatchPreparationTerminalExecutor(state), ) async for _ in preparation.run(state.task_text, stream=True, include_status_events=True): pass workflow = create_workflow( mode="manager_only", manager_answer=ManagerAnswerExecutor(adapter, state), seat_executors=[], approval_gate=object(), finalizer=object(), waiting_terminal=object(), seat_agent_ids=[], ) async for _ in workflow.run(state.task_text, stream=True, include_status_events=True): pass stages = [call.kwargs["execution_stage"] for call in adapter.run_seat_turn.await_args_list] assert stages == ["manager_dispatch", "direct_answer"] assert state.selected_seat_agent_ids == [] @pytest.mark.asyncio async def test_invalid_manager_plan_falls_back_to_clarification_without_starting_seats(): state = _state() state.manager_agent_id = "manager" state.thread_map["manager"] = "manager-thread" state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content="not a dispatch plan")), _emit_critical=AsyncMock(), ) ctx = _FakeCtx() await ManagerDispatchExecutor(adapter, state).handle("task", ctx) assert state.status == "waiting_input" assert state.intent_type == "clarification" assert state.selected_seat_agent_ids == [] assert ctx.messages[-1]["decision"] == "clarification" assert adapter.run_seat_turn.await_args.kwargs["execution_stage"] == "manager_dispatch" @pytest.mark.asyncio async def test_manager_dispatch_uses_capability_summary_and_selects_only_a_subset(): state = _state() state.manager_agent_id = "manager" state.thread_map["manager"] = "manager-thread" state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True state.capability_snapshot = { "agents": [ {"agent_id": "manager", "tools": [{"tool_name": "manager_write", "side_effect": "write", "risk_level": "R3"}]}, {"agent_id": "seat-a", "model_name": "m-a", "tools": [{"tool_name": "read_a", "side_effect": "read", "risk_level": "R1"}]}, {"agent_id": "seat-b", "model_name": "m-b", "tools": [{"tool_name": "read_b", "side_effect": "read", "risk_level": "R1"}]}, {"agent_id": "seat-c", "model_name": "m-c", "tools": [{"tool_name": "write_c", "side_effect": "write", "risk_level": "R3"}], "secret": "never-prompted"}, ] } decision = { "intent_type": "parallel_consultation", "selected_seat_agent_ids": ["seat-a", "seat-b"], "assignments": [ {"agent_id": "seat-a", "task_goal": "collect evidence", "delivery_requirements": ["sources"], "order": 1}, {"agent_id": "seat-b", "task_goal": "compare evidence", "delivery_requirements": ["gaps"], "order": 2}, ], "execution_mode": "parallel", "convergence_path": ["manager synthesis"], "reason": "two independent perspectives are needed", } adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content=json.dumps(decision))), _emit_critical=AsyncMock(), ) await ManagerDispatchExecutor(adapter, state).handle("task", _FakeCtx()) prompt = adapter.run_seat_turn.await_args.kwargs["message"] assert "trusted capability summary" in prompt.lower() assert "never-prompted" not in prompt assert adapter.run_seat_turn.await_args.kwargs["excluded_tools"] == ["manager_write"] assert state.selected_seat_agent_ids == ["seat-a", "seat-b"] assert state.plan["assignments"].keys() == {"seat-a", "seat-b"} @pytest.mark.asyncio @pytest.mark.parametrize("manager_result", [ "{\"intent_type\": \"unknown\"}", json.dumps({ "intent_type": "single_seat", "selected_seat_agent_ids": ["untrusted-seat"], "assignments": [{"agent_id": "untrusted-seat", "task_goal": "x", "delivery_requirements": ["y"], "order": 1}], "execution_mode": "single", "convergence_path": ["manager synthesis"], "reason": "invalid seat", }), ]) async def test_invalid_intent_or_seat_falls_back_to_clarification(manager_result): state = _state() state.manager_agent_id = "manager" state.thread_map["manager"] = "manager-thread" state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content=manager_result)), _emit_critical=AsyncMock(), ) await ManagerDispatchExecutor(adapter, state).handle("task", _FakeCtx()) assert state.status == "waiting_input" assert state.selected_seat_agent_ids == [] @pytest.mark.asyncio async def test_manager_dispatch_timeout_falls_back_to_clarification(): state = _state() state.manager_agent_id = "manager" state.thread_map["manager"] = "manager-thread" state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True adapter = SimpleNamespace( run_seat_turn=AsyncMock(side_effect=TimeoutError()), _emit_critical=AsyncMock(), ) await ManagerDispatchExecutor(adapter, state).handle("task", _FakeCtx()) assert state.status == "waiting_input" assert state.dispatch_plan["reason"] == "manager_dispatch_timeout" @pytest.mark.asyncio async def test_manager_dispatch_persists_and_binds_only_selected_seat_threads(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'dispatch-subset.db'}") async with engine.begin() as conn: import deerflow.persistence.maf_roundtable.model # noqa: F401 await conn.run_sync(Base.metadata.create_all) store = make_maf_roundtable_store(async_sessionmaker(engine, expire_on_commit=False)) session = await store.sessions.create_session({ "user_id": "u1", "task_text": "analyze", "mode": "structured", "manager_agent_id": "manager", "max_rounds": 8, }) await store.participants.replace_participants(session["id"], [{ "agent_id": "manager", "role": "manager", "sort_order": 0, }]) run = await store.runs.create_run(session["id"]) state = _state() state.session_id = session["id"] state.run_id = run["id"] state.thread_map = {"manager": "manager-thread"} state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True decision = { "intent_type": "parallel_consultation", "selected_seat_agent_ids": ["seat-a", "seat-b"], "assignments": [ {"agent_id": "seat-a", "task_goal": "evidence", "delivery_requirements": ["sources"], "order": 1}, {"agent_id": "seat-b", "task_goal": "review", "delivery_requirements": ["gaps"], "order": 2}, ], "execution_mode": "parallel", "convergence_path": ["manager synthesis"], "reason": "selected subset", } adapter = SimpleNamespace( _service=SimpleNamespace(store=store), run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content=json.dumps(decision))), ensure_threads=AsyncMock(return_value={"seat-a": "thread-a", "seat-b": "thread-b"}), _emit_critical=AsyncMock(), ) try: await ManagerDispatchExecutor(adapter, state).handle("task", _FakeCtx()) participants = await store.participants.list_participants(session["id"]) assert [row["agent_id"] for row in participants] == ["manager", "seat-a", "seat-b"] adapter.ensure_threads.assert_awaited_once() bound_agents = [row["agent_id"] for row in adapter.ensure_threads.await_args.args[2]] assert bound_agents == ["seat-a", "seat-b"] assert "seat-c" not in state.thread_map finally: await engine.dispose() @pytest.mark.asyncio async def test_group_chat_workflow_executes_verified_manager_subset_only(): pytest.importorskip("agent_framework") state = _state(["seat-a", "seat-b", "seat-c"]) state.mode = "group_chat" state.thread_map = {"manager": "manager-thread"} state.manager_agent_id = "manager" state.candidate_seat_agent_ids = ["seat-a", "seat-b", "seat-c"] state.dispatch_contract_enabled = True participant_rows = [{"agent_id": "manager", "role": "manager", "thread_id": "manager-thread"}] async def ensure_participants(_session_id, rows): participant_rows.extend(row for row in rows if row["agent_id"] not in {p["agent_id"] for p in participant_rows}) async def list_participants(_session_id): return list(participant_rows) dispatch = { "intent_type": "parallel_consultation", "selected_seat_agent_ids": ["seat-a", "seat-b"], "assignments": [ {"agent_id": "seat-a", "task_goal": "evidence", "delivery_requirements": ["sources"], "order": 1}, {"agent_id": "seat-b", "task_goal": "review", "delivery_requirements": ["gaps"], "order": 2}, ], "execution_mode": "group_chat", "convergence_path": ["manager synthesis"], "reason": "discussion requires two seats", } async def run_turn(**kwargs): if kwargs["execution_stage"] == "manager_dispatch": return _FakeSeatResponse(content=json.dumps(dispatch)) if kwargs["execution_stage"] == "finalizer": return _FakeSeatResponse(content="final") return _FakeSeatResponse(content=f"{kwargs['agent_id']} result") adapter = SimpleNamespace( _service=SimpleNamespace( store=SimpleNamespace( participants=SimpleNamespace( ensure_participants=ensure_participants, list_participants=list_participants, ), runs=SimpleNamespace(update_run=AsyncMock()), actions=SimpleNamespace(list_actions=AsyncMock(return_value=[])), ) ), run_seat_turn=AsyncMock(side_effect=run_turn), ensure_threads=AsyncMock(return_value={"seat-a": "thread-a", "seat-b": "thread-b"}), _emit_critical=AsyncMock(), ) preparation = build_manager_dispatch_workflow( manager_dispatch=ManagerDispatchExecutor(adapter, state), terminal=DispatchPreparationTerminalExecutor(state), ) async for _ in preparation.run(state.task_text, stream=True, include_status_events=True): pass workflow = create_workflow( mode="group_chat", coordinator=CoordinatorExecutor(adapter, state), seat_executors=[ SeatExecutor(adapter, state, agent_id) for agent_id in state.selected_seat_agent_ids or [] ], approval_gate=ApprovalGateExecutor(adapter, state, SimpleNamespace()), finalizer=FinalizerExecutor(adapter, state), waiting_terminal=WaitingTerminalExecutor(), seat_agent_ids=state.selected_seat_agent_ids or [], ) async for _ in workflow.run(state.task_text, stream=True, include_status_events=True): pass stages = [call.kwargs["execution_stage"] for call in adapter.run_seat_turn.await_args_list] assert stages.count("manager_dispatch") == 1 assert stages.count("group_chat") == 2 assert "seat-c" not in [call.kwargs["agent_id"] for call in adapter.run_seat_turn.await_args_list] assert [row["agent_id"] for row in participant_rows] == ["manager", "seat-a", "seat-b"] @pytest.mark.parametrize( ("execution_mode", "seat_count"), [("single", 1), ("parallel", 2), ("chain", 2)], ) def test_structured_factory_builds_each_trusted_topology(execution_mode, seat_count): pytest.importorskip("agent_framework") state = _state(["seat-a", "seat-b"][:seat_count]) adapter = SimpleNamespace(_emit_critical=AsyncMock()) seat_executors = [SeatExecutor(adapter, state, agent_id) for agent_id in state.seat_agent_ids] workflow = create_workflow( mode="structured", planner=PlannerExecutor(adapter, state), cross_review=CrossReviewExecutor(adapter, state), seat_executors=seat_executors, approval_gate=ApprovalGateExecutor(adapter, state, SimpleNamespace()), finalizer=FinalizerExecutor(adapter, state), waiting_terminal=WaitingTerminalExecutor(), seat_agent_ids=state.seat_agent_ids, execution_mode=execution_mode, ) assert workflow is not None @pytest.mark.asyncio async def test_planner_deterministic_assignments(): state = _state() adapter = SimpleNamespace(_emit_critical=AsyncMock()) planner = PlannerExecutor(adapter, state) ctx = _FakeCtx() await planner.handle("task", ctx) assert "planner" in state.completed_nodes assert state.plan is not None assert set(state.plan["assignments"]) == {"seat-a", "seat-b", "seat-c"} assert ctx.messages[-1]["decision"] == "parallel_seats" @pytest.mark.asyncio async def test_planner_skips_when_completed(): state = _state() state.mark_completed("planner") state.plan = {"source": "deterministic", "assignments": {}} adapter = SimpleNamespace(run_seat_turn=AsyncMock()) planner = PlannerExecutor(adapter, state) ctx = _FakeCtx() await planner.handle("task", ctx) adapter.run_seat_turn.assert_not_called() assert ctx.messages[-1].get("replayed") is True @pytest.mark.asyncio async def test_seat_partial_failure_does_not_raise(): state = _state(["seat-a", "seat-b"]) state.plan = { "assignments": { "seat-a": {"instruction": "do a", "role": "A"}, "seat-b": {"instruction": "do b", "role": "B"}, } } state.current_round = 1 async def _run(**kwargs): if kwargs["agent_id"] == "seat-a": raise RuntimeError("boom") return _FakeSeatResponse(content='{"summary":"ok-b","findings":["f"]}') adapter = SimpleNamespace( run_seat_turn=AsyncMock(side_effect=_run), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace(actions=SimpleNamespace(list_actions=AsyncMock(return_value=[])))), ) seat_a = SeatExecutor(adapter, state, "seat-a") seat_b = SeatExecutor(adapter, state, "seat-b") ctx_a, ctx_b = _FakeCtx(), _FakeCtx() await seat_a.handle({}, ctx_a) await seat_b.handle({}, ctx_b) assert state.partial_failure is True assert state.seat_outcomes["seat-a"]["status"] in ("failed", "error") assert state.seat_outcomes["seat-b"]["status"] == "completed" assert any(r.get("agent_id") == "seat-b" and r.get("content") for r in state.turn_results) assert "seat_seat-a" in state.completed_nodes assert "seat_seat-b" in state.completed_nodes @pytest.mark.asyncio async def test_seat_schema_one_repair_then_invalid(): state = _state(["seat-a"]) state.plan = {"assignments": {"seat-a": {"instruction": "x", "role": "A"}}} state.current_round = 1 calls = {"n": 0} async def _run(**kwargs): calls["n"] += 1 if calls["n"] == 1: return _FakeSeatResponse(content="not-json") return _FakeSeatResponse(content="still-bad") adapter = SimpleNamespace( run_seat_turn=AsyncMock(side_effect=_run), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace(actions=SimpleNamespace(list_actions=AsyncMock(return_value=[])))), ) seat = SeatExecutor(adapter, state, "seat-a") await seat.handle({}, _FakeCtx()) # 1 expert + 1 repair assert calls["n"] == 2 row = state.turn_results[-1] assert row["schema_status"] == "schema_invalid" @pytest.mark.asyncio async def test_proposed_not_executed_in_ledger(): state = _state(["seat-a"]) state.plan = {"assignments": {"seat-a": {"instruction": "x", "role": "A"}}} state.current_round = 1 content = ( '{"summary":"s","findings":["f"],' '"proposed_actions":[{"action_id":"p1","text":"write"}],' '"executed_actions":[{"action_id":"forge","text":"lie"}]}' ) adapter = SimpleNamespace( run_seat_turn=AsyncMock(return_value=_FakeSeatResponse(content=content)), _emit_critical=AsyncMock(), _service=SimpleNamespace(store=SimpleNamespace(actions=SimpleNamespace(list_actions=AsyncMock(return_value=[])))), ) await SeatExecutor(adapter, state, "seat-a").handle({}, _FakeCtx()) assert "p1" in proposed_action_ids(state.fact_ledger) assert "forge" not in executed_action_ids(state.fact_ledger) @pytest.mark.asyncio async def test_cross_review_barrier_marks_missing(): state = _state(["seat-a", "seat-b"]) state.seat_outcomes = {"seat-a": {"status": "completed"}} state.current_round = 1 adapter = SimpleNamespace(run_seat_turn=AsyncMock(), _emit_critical=AsyncMock()) review = CrossReviewExecutor(adapter, state) await review.handle({}, _FakeCtx()) assert state.seat_outcomes["seat-b"]["status"] == "missing" assert state.partial_failure is True assert "cross_review" in state.completed_nodes def test_structured_factory_requires_planner(): with pytest.raises(ValueError): create_workflow( mode="structured", seat_executors=[], approval_gate=object(), finalizer=object(), waiting_terminal=object(), seat_agent_ids=[], ) @pytest.mark.asyncio async def test_recover_skips_completed_seats(tmp_path, monkeypatch): """completed_nodes must not re-invoke run_seat_turn on recover.""" pytest.importorskip("agent_framework") from app.maf_roundtable.workflow_runner import WorkflowRunner engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'p3r.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) svc = make_maf_service(store) session = await svc.create_session( user_id="u1", body={ "task_text": "t", "mode": "structured", "participants": [ {"agent_id": "seat-a", "role": "seat"}, {"agent_id": "seat-b", "role": "seat"}, ], }, ) run = await svc.create_run(session_id=session["id"], user_id="u1") # Pretend planner + seat-a already finished; seat-b not yet. state = RoundtableState( session_id=session["id"], run_id=run["id"], task_text="t", mode="structured", model_name=None, reasoning_effort=None, max_rounds=8, write_policy="confirm_high_risk", manager_agent_id=None, seat_agent_ids=["seat-a", "seat-b"], thread_map={"seat-a": "ta", "seat-b": "tb"}, capability_snapshot=session.get("capability_snapshot"), seat_timeout_seconds=30, schema_repair_enabled=False, ) state.plan = { "source": "deterministic", "assignments": { "seat-a": {"instruction": "a", "role": "A"}, "seat-b": {"instruction": "b", "role": "B"}, }, } state.current_round = 1 state.stage = "experts" state.mark_completed("planner") state.mark_completed("seat_seat-a") state.seat_outcomes["seat-a"] = {"status": "completed"} state.turn_results.append({ "agent_id": "seat-a", "turn_index": 1, "content": '{"summary":"a"}', "tool_calls": [], "error": None, }) await store.runs.update_run(run["id"], {"status": "running", "maf_checkpoint": state.to_dict()}) calls: list[str] = [] async def _run_seat(**kwargs): aid = kwargs["agent_id"] calls.append(aid) return _FakeSeatResponse(content=f'{{"summary":"{aid}"}}') app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=svc, checkpointer=None)) runner = WorkflowRunner(app, service=svc, user_id="u1") monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat)) monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(return_value=state.thread_map)) monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock()) # Avoid needing a real SDK stream — drive recover path by patching _run_workflow # to invoke remaining executors with skip semantics. async def _fake_run_workflow(workflow, st): # Simulate graph: planner (skip) → seats → cross_review → approval → finalizer from app.maf_roundtable.workflow_executors import ( ApprovalGateExecutor, CrossReviewExecutor, FinalizerExecutor, PlannerExecutor, SeatExecutor, ) ctx = _FakeCtx() await PlannerExecutor(runner._adapter, st).handle("t", ctx) for aid in st.seat_agent_ids: await SeatExecutor(runner._adapter, st, aid).handle({}, _FakeCtx()) await CrossReviewExecutor(runner._adapter, st).handle({}, _FakeCtx()) await ApprovalGateExecutor(runner._adapter, st, svc).handle({}, _FakeCtx()) await FinalizerExecutor(runner._adapter, st).handle({}, _FakeCtx()) st.status = "completed" monkeypatch.setattr(runner, "_run_workflow", _fake_run_workflow) try: await runner.recover(run["id"]) # seat-a was completed — must not be called; seat-b should run once. assert "seat-a" not in calls assert calls.count("seat-b") == 1 finally: await engine.dispose()