698 lines
27 KiB
Python
698 lines
27 KiB
Python
"""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()
|