1320 lines
51 KiB
Python
1320 lines
51 KiB
Python
"""Phase 2 P1 acceptance tests for the MAF roundtable workflow execution.
|
|
|
|
These tests exercise the REAL MAF SDK workflow orchestration (WorkflowBuilder,
|
|
Executor subclasses, streaming events, checkpoint persistence) with a mocked
|
|
DeerFlow runtime client (no real LLM calls). They verify the four completion
|
|
criteria for Phase 2 P1 closure:
|
|
|
|
1. Real MAF Workflow is running.
|
|
2. Write tools never execute before approval (interception before execution).
|
|
3. After approval, the workflow resumes and executes exactly once (at-most-once).
|
|
4. Cancel and restart recovery don't leave DeerFlow orphan tasks.
|
|
|
|
Additionally covers:
|
|
- Two seats each create DeerFlow threads owned by the session user.
|
|
- MAF group_chat and structured both run a real SDK path.
|
|
- Secrets (token/cookie/password/long secret) never appear in events/errors/logs.
|
|
- source_event_id dedup at DB level.
|
|
- Audit event write failure →external write does not execute (run fails).
|
|
- ActionCoordinator: encrypted params never returned from API/events/logs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import itertools
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.action_coordinator import (
|
|
ActionCoordinator,
|
|
classify_tool,
|
|
decrypt_params,
|
|
encrypt_params,
|
|
tool_needs_approval,
|
|
)
|
|
from app.maf_roundtable.deerflow_agent_adapter import DeerFlowAgentAdapter
|
|
from app.maf_roundtable.runtime_client import ExecutionPrincipal
|
|
from app.maf_roundtable.workflow_runner import WorkflowRunner
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
|
|
# ── fixtures ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_accept.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)
|
|
yield make_maf_roundtable_store(sf)
|
|
await engine.dispose()
|
|
|
|
|
|
def _svc(store):
|
|
from maf_test_helpers import make_maf_service
|
|
|
|
return make_maf_service(store, enabled=True, admin_diagnostic_access=False)
|
|
|
|
|
|
def _make_app(store):
|
|
"""Build a minimal fake app with the state the runner needs."""
|
|
app = SimpleNamespace()
|
|
app.state = SimpleNamespace()
|
|
app.state.maf_roundtable_service = _svc(store)
|
|
app.state.checkpointer = AsyncMock()
|
|
app.state.thread_store = AsyncMock()
|
|
app.state.run_manager = AsyncMock()
|
|
app.state.stream_bridge = None
|
|
return app
|
|
|
|
|
|
class _FakeRunRecord:
|
|
"""Mimics the RunRecord returned by start_run."""
|
|
|
|
def __init__(self, run_id="df-run-1"):
|
|
self.run_id = run_id
|
|
self.task = asyncio.ensure_future(asyncio.sleep(0))
|
|
|
|
|
|
def _patch_runtime_client(adapter: DeerFlowAgentAdapter, *, responses: dict[str, str] | None = None, tool_calls_map: dict[str, list] | None = None):
|
|
"""Patch the adapter's runtime client to return controlled responses.
|
|
|
|
``responses`` maps agent_id →content text.
|
|
``tool_calls_map`` maps agent_id →list of tool call dicts the AI "generated".
|
|
"""
|
|
responses = responses or {}
|
|
tool_calls_map = tool_calls_map or {}
|
|
call_log: list[dict] = []
|
|
write_tool_counter: dict[str, int] = {"count": 0}
|
|
|
|
async def fake_create_thread(principal: ExecutionPrincipal) -> str:
|
|
return f"thread-{principal.session_id[:8]}-{len(call_log)}"
|
|
|
|
async def fake_start_agent_run(principal, *, thread_id, agent_id, message, excluded_tools=None, approved_tool=None, turn_index=0, capability_snapshot=None, write_policy="confirm_high_risk", **kwargs):
|
|
call_log.append({
|
|
"agent_id": agent_id,
|
|
"thread_id": thread_id,
|
|
"message": message[:200],
|
|
"excluded_tools": list(excluded_tools or []),
|
|
"approved_tool": approved_tool,
|
|
"turn_index": turn_index,
|
|
"capability_snapshot": capability_snapshot,
|
|
"write_policy": write_policy,
|
|
})
|
|
|
|
# Phase-2 simulation: when this is a normal seat turn (no approved_tool),
|
|
# the WriteGuard middleware would intercept write tools inside the real
|
|
# DeerFlow run and create pending actions. Since the fake bypasses the
|
|
# middleware, we replay that here so the action/approval get created
|
|
# exactly as the real middleware would.
|
|
if not approved_tool and write_policy != "auto":
|
|
from app.maf_roundtable.action_coordinator import ActionCoordinator
|
|
|
|
tcs = tool_calls_map.get(agent_id, [])
|
|
coord = ActionCoordinator(adapter._service, user_id=principal.user_id)
|
|
try:
|
|
await coord.intercept_write_tool_calls(
|
|
run_id=principal.run_id,
|
|
session_id=principal.session_id,
|
|
agent_id=agent_id,
|
|
turn_index=turn_index,
|
|
tool_calls=tcs,
|
|
capability_snapshot=capability_snapshot,
|
|
write_policy=write_policy,
|
|
)
|
|
except Exception:
|
|
pass # best-effort replay for the test harness
|
|
|
|
return _FakeRunRecord(run_id=f"df-{agent_id}")
|
|
|
|
async def fake_read_messages(principal, thread_id):
|
|
agent_id = ""
|
|
for c in call_log:
|
|
if c["thread_id"] == thread_id:
|
|
agent_id = c["agent_id"]
|
|
break
|
|
|
|
# Phase-2 semantics: the WriteGuard middleware intercepts write tools
|
|
# inside the DeerFlow run. A normal seat turn (no ``approved_tool``)
|
|
# means the write was intercepted →counter stays put. Only the
|
|
# ToolExecutor's deterministic run (via execute_tool_directly) sets
|
|
# the write counter.
|
|
tcs = tool_calls_map.get(agent_id, [])
|
|
content = responses.get(agent_id, f"Response from {agent_id}")
|
|
messages = [{"type": "ai", "content": content, "usage_metadata": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}}]
|
|
if tcs:
|
|
messages[0]["tool_calls"] = tcs
|
|
return messages
|
|
|
|
async def fake_execute_tool_directly(*, thread_id, agent_id, tool_name, params, principal, **kwargs):
|
|
"""Simulate direct tool execution —the LLM is bypassed (P0-2).
|
|
Increment the write counter for the approved tool. Extra P0-1 context
|
|
kwargs (turn_index/capability_snapshot/— are accepted and ignored."""
|
|
write_tool_counter["count"] += 1
|
|
return {"status": "succeeded", "result": {"tool": tool_name, "params": params}}
|
|
|
|
adapter._client.create_thread = fake_create_thread
|
|
adapter._client.start_agent_run = fake_start_agent_run
|
|
adapter._client.read_thread_messages = fake_read_messages
|
|
adapter._client.cancel_run = AsyncMock(return_value=True)
|
|
adapter._client.execute_tool_directly = fake_execute_tool_directly
|
|
return call_log, write_tool_counter
|
|
|
|
|
|
# ── 1. Real SDK structured workflow, 2 seats ─────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_sdk_structured_workflow_two_seats(store):
|
|
"""Build and run a real MAF SDK structured workflow with 2 seat executors."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "分析市场趋势",
|
|
"mode": "structured",
|
|
"participants": [
|
|
{"agent_id": "analyst-a", "role": "seat"},
|
|
{"agent_id": "analyst-b", "role": "seat"},
|
|
],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(runner.adapter, responses={
|
|
"analyst-a": "市场呈上升趋势",
|
|
"analyst-b": "需注意风险因素",
|
|
})
|
|
|
|
await runner.execute(s, run)
|
|
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "completed"
|
|
assert final_run["finished_at"] is not None
|
|
assert final_run["started_at"] is not None
|
|
|
|
final_session = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert final_session["status"] == "completed"
|
|
assert final_session["final_result"] is not None
|
|
|
|
|
|
# ── 2. Two seats create threads owned by session user ──────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_seats_create_threads_owned_by_session_user(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"participants": [
|
|
{"agent_id": "seat-1", "role": "seat"},
|
|
{"agent_id": "seat-2", "role": "seat"},
|
|
],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
|
|
created_threads: list[dict] = []
|
|
orig_create = runner.adapter._client.create_thread
|
|
|
|
async def tracking_create(principal: ExecutionPrincipal):
|
|
tid = await orig_create(principal)
|
|
created_threads.append({"thread_id": tid, "user_id": principal.user_id, "session_id": principal.session_id})
|
|
return tid
|
|
|
|
runner.adapter._client.create_thread = tracking_create
|
|
_patch_runtime_client(runner.adapter)
|
|
runner.adapter._client.create_thread = tracking_create
|
|
|
|
await runner.execute(s, run)
|
|
|
|
assert len(created_threads) == 2
|
|
assert all(t["user_id"] == "u1" for t in created_threads)
|
|
assert all(t["session_id"] == s["id"] for t in created_threads)
|
|
assert created_threads[0]["thread_id"] != created_threads[1]["thread_id"]
|
|
|
|
|
|
# ── 3. Two seats execute in order, producing events ─────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_seats_execute_in_order_producing_events(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"participants": [
|
|
{"agent_id": "seat-x", "role": "seat"},
|
|
{"agent_id": "seat-y", "role": "seat"},
|
|
],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(runner.adapter)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
|
|
event_types = [e["event_type"] for e in events]
|
|
|
|
assert event_types.count("agent.turn_started") >= 2
|
|
assert event_types.count("agent.message.completed") >= 2
|
|
assert "run.completed" in event_types
|
|
|
|
|
|
# ── 4. Seat tool call + skill + artifact events ─────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seat_tool_call_and_skill_artifact_events(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"participants": [{"agent_id": "tool-seat", "role": "seat"}],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
|
|
from app.maf_roundtable.event_normalizer import (
|
|
make_artifact_created,
|
|
make_skill_completed,
|
|
make_skill_started,
|
|
)
|
|
|
|
await runner.adapter._emit_critical(run["id"], {
|
|
"event_type": "tool.requested",
|
|
"agent_id": "tool-seat",
|
|
"turn_index": 1,
|
|
"payload": {"tool_name": "web_search", "tool_call_id": "tc1"},
|
|
})
|
|
await runner.adapter._emit_critical(run["id"], {
|
|
"event_type": "tool.completed",
|
|
"agent_id": "tool-seat",
|
|
"turn_index": 1,
|
|
"payload": {"tool_name": "web_search", "tool_call_id": "tc1", "status": "completed"},
|
|
})
|
|
await runner.adapter._emit_critical(run["id"], make_skill_started("tool-seat", 1, skill_name="data-analysis"))
|
|
await runner.adapter._emit_critical(run["id"], make_skill_completed("tool-seat", 1, skill_name="data-analysis", result_preview="分析完成"))
|
|
await runner.adapter._emit_critical(run["id"], make_artifact_created("tool-seat", 1, artifact_type="chart", artifact_id="art-1", title="趋势图"))
|
|
|
|
events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
|
|
types = [e["event_type"] for e in events]
|
|
assert "tool.requested" in types
|
|
assert "tool.completed" in types
|
|
assert "skill.started" in types
|
|
assert "skill.completed" in types
|
|
assert "artifact.created" in types
|
|
|
|
|
|
# ── 5. P1: Write tool counter —0 before approval, 1 after, still 1 after repeated resume ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_tool_counter_zero_before_approval(store):
|
|
"""P1: write tools are intercepted before execution. The write tool
|
|
counter must be 0 before any approval."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "分析完成，需要创建工单"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
# Write tool was excluded →never executed →counter is 0.
|
|
assert write_counter["count"] == 0, "write tool executed before approval!"
|
|
|
|
# An action + approval should have been created.
|
|
actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
assert len(actions) == 1
|
|
assert actions[0]["tool_name"] == "create_ticket"
|
|
assert actions[0]["status"] == "waiting_approval"
|
|
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
assert len(approvals) == 1
|
|
assert approvals[0]["status"] == "pending"
|
|
|
|
# Run should be in waiting_approval (not completed).
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "waiting_approval"
|
|
# waiting_approval must NOT write finished_at.
|
|
assert final_run["finished_at"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_tool_counter_one_after_approval(store):
|
|
"""P1: after approval, the workflow resumes and the write tool executes
|
|
exactly once."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
# Execute the workflow →creates action + approval →waiting_approval.
|
|
await runner.execute(s, run)
|
|
|
|
assert write_counter["count"] == 0 # Before approval: 0.
|
|
|
|
# Approve.
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
approval = approvals[0]
|
|
decided = await svc.decide_approval(approval_id=approval["id"], decision="approved", user_id="u1")
|
|
assert decided["status"] == "approved"
|
|
|
|
# Resume the workflow.
|
|
resumed = await runner.resume(run["id"])
|
|
assert resumed is not None
|
|
assert resumed["status"] == "completed"
|
|
|
|
# After approval + resume: write tool executed exactly once.
|
|
assert write_counter["count"] == 1, "write tool should execute exactly once after approval"
|
|
|
|
# Action should be succeeded.
|
|
actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
assert actions[0]["status"] == "succeeded"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_tool_still_one_after_repeated_resume(store):
|
|
"""P1: repeated resume does not double-execute the write tool."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
|
|
|
|
# First resume.
|
|
await runner.resume(run["id"])
|
|
assert write_counter["count"] == 1
|
|
|
|
# Second resume —should not re-execute.
|
|
await runner.resume(run["id"])
|
|
assert write_counter["count"] == 1, "repeated resume must not double-execute"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_resume_second_sees_running_tool_runs_once(store):
|
|
"""P0-3: two concurrent resumes. The first wins the run-level CAS and blocks
|
|
inside the tool; the second sees the run already ``running`` and returns
|
|
WITHOUT writing a terminal status. The tool executes exactly once; only after
|
|
the first resume's tool is released does the run finalize (report) →completed.
|
|
"""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
# Replace the direct-tool executor with a blocking one so we can interleave
|
|
# a second resume while the first is mid-execution.
|
|
write_counter = {"count": 0}
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def blocking_execute_tool_directly(*, thread_id, agent_id, tool_name, params, principal, **kwargs):
|
|
write_counter["count"] += 1
|
|
started.set() # signal: first resume has claimed run+action and is executing
|
|
await release.wait() # block until the test releases it
|
|
return {"status": "succeeded", "result": {"tool": tool_name}}
|
|
|
|
runner.adapter._client.execute_tool_directly = blocking_execute_tool_directly
|
|
|
|
# Drive the workflow to waiting_approval, then approve.
|
|
await runner.execute(s, run)
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
|
|
|
|
# First resume (background) —wins the run CAS, then blocks in the tool.
|
|
first_task = asyncio.ensure_future(runner.resume(run["id"]))
|
|
await asyncio.wait_for(started.wait(), timeout=5)
|
|
assert write_counter["count"] == 1, "tool must have started exactly once"
|
|
|
|
# Second resume —the run is already 'running'; it must return that and must
|
|
# NOT write a terminal status or re-execute the tool.
|
|
second = await runner.resume(run["id"])
|
|
assert second is not None and second["status"] == "running", \
|
|
f"second concurrent resume must see running, got {second and second['status']}"
|
|
assert write_counter["count"] == 1, "second resume must not re-execute the tool"
|
|
|
|
# Release the first resume's tool →it finalizes (report) →completed.
|
|
release.set()
|
|
first = await asyncio.wait_for(first_task, timeout=10)
|
|
assert first is not None and first["status"] == "completed"
|
|
assert write_counter["count"] == 1, "tool executes exactly once across both resumes"
|
|
|
|
# The completed run carries a synthesized final report (finalizer re-entered).
|
|
final_run = await svc.store.runs.get_run(run["id"])
|
|
assert final_run["finished_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_with_missing_secret_fails_without_executing(store):
|
|
"""P0-2: if the action secret is deleted before resume, the write tool must
|
|
execute **0 times**, the action is ``failed`` (not succeeded), and the parent
|
|
run is ``failed`` —never completed, never a retry of the external write."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
|
|
|
|
# Delete the encrypted secret out from under the approved action.
|
|
actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
action_id = actions[0]["id"]
|
|
await store.action_secrets.delete_by_action(action_id)
|
|
|
|
resumed = await runner.resume(run["id"])
|
|
|
|
assert write_counter["count"] == 0, "tool must NOT execute when the secret is gone"
|
|
final_actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
assert final_actions[0]["status"] == "failed"
|
|
assert final_actions[0]["status"] != "succeeded"
|
|
assert resumed is not None and resumed["status"] == "failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_with_corrupt_secret_fails_without_executing(store):
|
|
"""P0-2: if the action secret cannot be decrypted (wrong key / garbage),
|
|
resume fails the action and the run with the tool executing **0 times**."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"}],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
|
|
|
|
# Replace the ciphertext with garbage that Fernet cannot decrypt.
|
|
actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
action_id = actions[0]["id"]
|
|
await store.action_secrets.delete_by_action(action_id)
|
|
await store.action_secrets.create(action_id=action_id, arguments_encrypted="corrupt-token")
|
|
|
|
resumed = await runner.resume(run["id"])
|
|
|
|
assert write_counter["count"] == 0, "tool must NOT execute when the secret is undecryptable"
|
|
final_actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
assert final_actions[0]["status"] == "failed"
|
|
assert resumed is not None and resumed["status"] == "failed"
|
|
|
|
|
|
# ── 6. P1: Rejected approval →write count stays 0 ──────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejected_approval_write_count_stays_zero(store):
|
|
"""P1: after a rejected approval, the external write count stays 0 and
|
|
the action is marked rejected."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_, write_counter = _patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "需要创建工单"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="rejected", user_id="u1")
|
|
|
|
await runner.resume(run["id"])
|
|
|
|
assert write_counter["count"] == 0, "rejected approval must not execute write tool"
|
|
|
|
actions = await svc.list_actions(run_id=run["id"], user_id="u1")
|
|
assert actions[0]["status"] == "rejected"
|
|
|
|
|
|
# ── 7. P1: In-process cancel cascades to DeerFlow ───────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_in_process_cancel_cascades_to_deerflow(store):
|
|
"""P1: cancel calls DeerFlow run_manager.cancel on all active seat runs."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "t", "participants": [{"agent_id": "seat-1", "role": "seat"}]},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
# Create a seat run with a DeerFlow run id.
|
|
seat_run = await store.seat_runs.create_seat_run(
|
|
maf_run_id=run["id"], agent_id="seat-1", turn_index=1, deerflow_thread_id="thread-x"
|
|
)
|
|
await store.seat_runs.mark_running(seat_run["id"], deerflow_run_id="df-run-42")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
runner.adapter._client.cancel_run = AsyncMock(return_value=True)
|
|
|
|
result = await runner.cancel(run["id"])
|
|
assert result["all_cancelled"] is True
|
|
|
|
# DeerFlow cancel_run was called with the right run id.
|
|
runner.adapter._client.cancel_run.assert_called_once_with("df-run-42")
|
|
|
|
# Seat run should be cancelled.
|
|
seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
assert seat_runs[0]["status"] == "cancelled"
|
|
|
|
|
|
# ── 8. P1: Simulated restart cancel via persisted seat runs ──────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restart_cancel_via_persisted_seat_runs(store):
|
|
"""P1: after a simulated restart, the system can still cancel outstanding
|
|
DeerFlow runs via persisted seat_run records."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "t", "participants": [{"agent_id": "long-seat", "role": "seat"}]},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
seat_run = await store.seat_runs.create_seat_run(
|
|
maf_run_id=run["id"], agent_id="long-seat", turn_index=1, deerflow_thread_id="thread-x"
|
|
)
|
|
await store.seat_runs.mark_running(seat_run["id"], deerflow_run_id="df-run-99")
|
|
|
|
# "Restart" —build a fresh runner (new adapter, new client).
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
runner.adapter._client.cancel_run = AsyncMock(return_value=True)
|
|
|
|
cancelled = await runner.cancel(run["id"])
|
|
assert cancelled["all_cancelled"] is True
|
|
runner.adapter._client.cancel_run.assert_called_once_with("df-run-99")
|
|
|
|
seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
assert seat_runs[0]["status"] == "cancelled"
|
|
|
|
|
|
# ── 9. P1: Two-phase cancel —cancelling then cancelled ─────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_phase_cancel_cancelling_then_cancelled(store):
|
|
"""P1: cancel first marks seat runs as cancelling, then cancelled after
|
|
DeerFlow cancel succeeds."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
sr = await store.seat_runs.create_seat_run(
|
|
maf_run_id=run["id"], agent_id="s1", turn_index=1, deerflow_thread_id="t1"
|
|
)
|
|
await store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")
|
|
|
|
infos = await store.seat_runs.cancel_all_active(run["id"])
|
|
assert len(infos) == 1
|
|
assert infos[0]["deerflow_run_id"] == "df-1"
|
|
|
|
runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
assert runs[0]["status"] == "cancelling"
|
|
|
|
await store.seat_runs.mark_cancelled(sr["id"])
|
|
runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
assert runs[0]["status"] == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_failed_when_deerflow_cancel_fails(store):
|
|
"""P1: when DeerFlow cancel fails, the seat run is marked cancel_failed."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
sr = await store.seat_runs.create_seat_run(
|
|
maf_run_id=run["id"], agent_id="s1", turn_index=1, deerflow_thread_id="t1"
|
|
)
|
|
await store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
runner.adapter._client.cancel_run = AsyncMock(return_value=False)
|
|
|
|
result = await runner.cancel(run["id"])
|
|
assert result["all_cancelled"] is False
|
|
|
|
runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
assert runs[0]["status"] == "cancel_failed"
|
|
|
|
|
|
# ── 10. MAF group_chat runs a real SDK path ──────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_chat_runs_real_sdk_path(store):
|
|
"""P1: group_chat mode runs a real MAF SDK workflow path."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "讨论方案",
|
|
"mode": "group_chat",
|
|
"max_rounds": 2,
|
|
"participants": [
|
|
{"agent_id": "g-seat-1", "role": "seat"},
|
|
{"agent_id": "g-seat-2", "role": "seat"},
|
|
],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(runner.adapter, responses={
|
|
"g-seat-1": "我认为方案A更好",
|
|
"g-seat-2": "我同意",
|
|
})
|
|
|
|
await runner.execute(s, run)
|
|
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "completed"
|
|
|
|
|
|
# ── 11. Audit event failure prevents external write ──────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_event_failure_prevents_external_write(store):
|
|
"""P1: when a critical audit event fails to persist, the adapter raises
|
|
AuditEventError and the run fails —external writes never execute."""
|
|
from app.maf_roundtable.deerflow_agent_adapter import AuditEventError
|
|
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
|
|
with patch.object(svc, "append_event", side_effect=RuntimeError("DB write failed")):
|
|
with pytest.raises(AuditEventError, match="failed to persist"):
|
|
await runner.adapter._emit_critical(run["id"], {
|
|
"event_type": "tool.requested",
|
|
"agent_id": "seat",
|
|
"turn_index": 1,
|
|
"payload": {"tool_name": "write_data"},
|
|
})
|
|
|
|
|
|
# ── 12. DeerFlow task error marks run failed ─────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deerflow_task_error_marks_run_failed(store):
|
|
"""P3: when a DeerFlow seat turn raises, the seat is soft-failed
|
|
(``partial_failure``) and the structured run still completes degraded;
|
|
secrets must not appear in persisted seat / run errors."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "t", "participants": [{"agent_id": "boom-seat", "role": "seat"}]},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
|
|
async def exploding_start(*args, **kwargs):
|
|
raise RuntimeError("LLM provider unavailable: token=sk-secret-12345")
|
|
|
|
runner.adapter._client.create_thread = AsyncMock(return_value="thread-1")
|
|
runner.adapter._client.start_agent_run = exploding_start
|
|
|
|
await runner.execute(s, run)
|
|
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "completed"
|
|
assert final_run["finished_at"] is not None
|
|
ckpt = final_run.get("maf_checkpoint") or {}
|
|
assert ckpt.get("partial_failure") is True
|
|
outcome = (ckpt.get("seat_outcomes") or {}).get("boom-seat") or {}
|
|
assert outcome.get("status") in ("failed", "error")
|
|
persisted = " ".join(
|
|
[
|
|
str(final_run.get("error_message_safe") or ""),
|
|
str(outcome.get("error") or ""),
|
|
str(ckpt),
|
|
]
|
|
)
|
|
assert "sk-secret-12345" not in persisted
|
|
|
|
|
|
# ── 13. Secrets never in events / errors / logs ──────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_secrets_never_in_events_or_errors(store):
|
|
"""P1: tokens, cookies, passwords, and long secrets are stripped from
|
|
event payloads, error messages, and seat-run errors."""
|
|
from app.maf_roundtable.sanitize import sanitize
|
|
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
secret_payload = {
|
|
"tool_name": "api_call",
|
|
"authorization": "Bearer sk-live-abcdef123456789",
|
|
"cookie": "session=abc123; token=xyz789",
|
|
"password": "hunter2",
|
|
"api_key": "sk-proj-1234567890abcdef",
|
|
"nested": {"secret": "deep-secret-value", "data": "safe"},
|
|
"long_text": "x" * 600 + " token=embedded-secret-999 " + "y" * 600,
|
|
}
|
|
event = await svc.append_event(
|
|
run_id=run["id"], user_id="u1", event_type="tool.completed", payload=secret_payload
|
|
)
|
|
|
|
stored = event["payload"]
|
|
assert stored["authorization"] == "***REDACTED***"
|
|
assert stored["cookie"] == "***REDACTED***"
|
|
assert stored["password"] == "***REDACTED***"
|
|
assert stored["api_key"] == "***REDACTED***"
|
|
assert stored["nested"]["secret"] == "***REDACTED***"
|
|
assert stored["nested"]["data"] == "safe"
|
|
assert "embedded-secret-999" not in stored["long_text"]
|
|
|
|
long_secret = "prefix " * 200 + "password=super-secret-value " + "suffix " * 200
|
|
sanitized = sanitize(long_secret)
|
|
assert "super-secret-value" not in sanitized
|
|
|
|
|
|
# ── 14. source_event_id dedup at DB level ────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_event_id_dedup_at_db_level(store):
|
|
"""P1: events with the same (run_id, source_event_id) are deduped at the DB level."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
e1 = await svc.append_event(
|
|
run_id=run["id"], user_id="u1", event_type="agent.message.delta",
|
|
payload={"text": "hello"}, source_event_id="msg-abc-123",
|
|
)
|
|
e2 = await svc.append_event(
|
|
run_id=run["id"], user_id="u1", event_type="agent.message.delta",
|
|
payload={"text": "hello again"}, source_event_id="msg-abc-123",
|
|
)
|
|
|
|
assert e1["id"] == e2["id"]
|
|
assert e1["seq"] == e2["seq"]
|
|
|
|
events, latest = await svc.list_events(run_id=run["id"], user_id="u1")
|
|
assert len(events) == 1
|
|
assert latest == 1
|
|
|
|
|
|
# ── 15. ActionCoordinator: encrypted params never returned ───────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_encrypted_action_params_never_returned(store):
|
|
"""P1: encrypted actual execution params are never returned from the
|
|
actions API, events, or logs."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
coord = ActionCoordinator(svc, user_id="u1")
|
|
|
|
# Create an action with encrypted params.
|
|
action = await store.actions.create_action({
|
|
"session_id": s["id"],
|
|
"run_id": run["id"],
|
|
"agent_id": "writer",
|
|
"tool_name": "create_ticket",
|
|
"idempotency_key": f"{run['id']}:writer:1:tc-1",
|
|
"risk_level": "R3",
|
|
"status": "waiting_approval",
|
|
"arguments": {"title": "test", "authorization": "***REDACTED***"},
|
|
})
|
|
await store.action_secrets.create(
|
|
action_id=action["id"],
|
|
arguments_encrypted=encrypt_params({"title": "test", "password": "real-secret-xyz"}),
|
|
)
|
|
|
|
# The action API response must NOT contain encrypted params.
|
|
api_action = await svc.get_action(action_id=action["id"], user_id="u1")
|
|
assert "arguments_encrypted" not in str(api_action).lower() or "arguments_encrypted" not in api_action
|
|
assert "password" not in str(api_action.get("arguments", {}))
|
|
|
|
# The encrypted params can only be decrypted server-side.
|
|
decrypted = await coord.get_approved_action_params(action["id"])
|
|
assert decrypted is not None
|
|
assert decrypted["password"] == "real-secret-xyz"
|
|
|
|
# But the decrypted params must never appear in the action's API response.
|
|
api_action2 = await svc.get_action(action_id=action["id"], user_id="u1")
|
|
assert "real-secret-xyz" not in json.dumps(api_action2)
|
|
|
|
|
|
# ── 16. ActionCoordinator: tool classification ───────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_classification_metadata_driven(store):
|
|
"""P1: tools are classified from capability snapshot metadata, not name
|
|
prefixes. Tools without metadata use safe default (write/R3 →needs approval)."""
|
|
snapshot = {
|
|
"tools": [
|
|
{"name": "search", "side_effect": "read", "risk_level": "R0"},
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
{"name": "unknown_tool"}, # no metadata →safe default
|
|
],
|
|
}
|
|
|
|
# Read tool →no approval needed.
|
|
assert classify_tool("search", snapshot) == ("read", "R0")
|
|
assert not tool_needs_approval("search", "confirm_high_risk", snapshot)
|
|
|
|
# Write tool R3 →approval needed in confirm_high_risk.
|
|
assert classify_tool("create_ticket", snapshot) == ("write", "R3")
|
|
assert tool_needs_approval("create_ticket", "confirm_high_risk", snapshot)
|
|
|
|
# Unknown tool →safe default (write/R3) →approval needed.
|
|
assert classify_tool("unknown_tool", snapshot) == ("write", "R3")
|
|
assert tool_needs_approval("unknown_tool", "confirm_high_risk", snapshot)
|
|
|
|
# In auto mode →no approval needed.
|
|
assert not tool_needs_approval("create_ticket", "auto", snapshot)
|
|
|
|
# In confirm_all_writes →any write needs approval.
|
|
assert tool_needs_approval("create_ticket", "confirm_all_writes", snapshot)
|
|
|
|
# No snapshot →safe default.
|
|
assert classify_tool("any_tool", None) == ("write", "R3")
|
|
|
|
|
|
# ── 17. ActionCoordinator: encryption round-trip ─────────────────────────
|
|
|
|
|
|
def test_encryption_round_trip():
|
|
"""P1: encrypted params can be decrypted back correctly."""
|
|
params = {"title": "test", "password": "secret123", "nested": {"key": "val"}}
|
|
encrypted = encrypt_params(params)
|
|
decrypted = decrypt_params(encrypted)
|
|
assert decrypted == params
|
|
# The ciphertext must not contain the plaintext.
|
|
assert "secret123" not in encrypted
|
|
assert "test" not in encrypted
|
|
|
|
|
|
# ── 18. waiting_approval does NOT write finished_at ──────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_waiting_approval_no_finished_at(store):
|
|
"""P1: waiting_approval is NOT a terminal state —finished_at must not
|
|
be written, and both session and run should be in waiting_approval."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "需要创建工单"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "waiting_approval"
|
|
assert final_run["finished_at"] is None
|
|
|
|
final_session = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert final_session["status"] == "waiting_approval"
|
|
|
|
|
|
# ── 19. RoundtableState.from_dict is actually called by resume ────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_uses_from_dict(store):
|
|
"""P1: RoundtableState.from_dict() is actually called by the resume path,
|
|
not left idle."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "t",
|
|
"write_policy": "confirm_all_writes",
|
|
"participants": [{"agent_id": "writer", "role": "seat"}],
|
|
"capability_snapshot": {
|
|
"agents": [{"agent_id": "writer"}],
|
|
"tools": [
|
|
{"name": "create_ticket", "side_effect": "write", "risk_level": "R3"},
|
|
],
|
|
},
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
_patch_runtime_client(
|
|
runner.adapter,
|
|
responses={"writer": "工单已创建"},
|
|
tool_calls_map={"writer": [{"name": "create_ticket", "args": {"title": "test"}, "id": "tc-1"}]},
|
|
)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
# Verify checkpoint was persisted.
|
|
run_after = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert run_after["maf_checkpoint"] is not None
|
|
assert run_after["maf_checkpoint"]["run_id"] == run["id"]
|
|
|
|
# Approve and resume.
|
|
approvals = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
await svc.decide_approval(approval_id=approvals[0]["id"], decision="approved", user_id="u1")
|
|
resumed = await runner.resume(run["id"])
|
|
assert resumed is not None
|
|
assert resumed["status"] == "completed"
|
|
|
|
|
|
# ── 20. Token usage aggregation ──────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_usage_aggregated_on_completion(store):
|
|
from app.maf_roundtable.workflow_runner import _aggregate_token_usage
|
|
|
|
turns = [
|
|
{"agent_id": "a", "turn_index": 1, "usage": {"input_tokens": 100, "output_tokens": 50, "total_tokens": 150}},
|
|
{"agent_id": "b", "turn_index": 1, "usage": {"input_tokens": 200, "output_tokens": 80, "total_tokens": 280}},
|
|
{"agent_id": "a", "turn_index": 2, "usage": None},
|
|
]
|
|
agg = _aggregate_token_usage(turns)
|
|
assert agg["total_input_tokens"] == 300
|
|
assert agg["total_output_tokens"] == 130
|
|
assert agg["total_tokens"] == 430
|
|
assert len(agg["per_seat"]) == 2
|
|
|
|
|
|
# ── 21. Capability snapshot reading ──────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_snapshot_drives_seat_config(store):
|
|
from app.maf_roundtable.workflow_executors import _agent_capabilities_from_snapshot
|
|
|
|
snapshot = {
|
|
"agents": [
|
|
{
|
|
"agent_id": "seat-1",
|
|
"model_name": "gpt-4o",
|
|
"reasoning_effort": "high",
|
|
"thinking_enabled": False,
|
|
"subagent_enabled": True,
|
|
"excluded_tools": ["dangerous_tool"],
|
|
"memory_recall_disabled": True,
|
|
"memory_injection_enabled": False,
|
|
},
|
|
],
|
|
}
|
|
caps = _agent_capabilities_from_snapshot(snapshot, "seat-1")
|
|
assert caps["model_name"] == "gpt-4o"
|
|
assert caps["reasoning_effort"] == "high"
|
|
assert caps["thinking_enabled"] is False
|
|
assert caps["subagent_enabled"] is True
|
|
assert caps["excluded_tools"] == ["dangerous_tool"]
|
|
assert caps["memory_recall_disabled"] is True
|
|
assert caps["memory_injection_enabled"] is False
|
|
|
|
caps2 = _agent_capabilities_from_snapshot(snapshot, "unknown-agent")
|
|
assert caps2 == {}
|
|
|
|
caps3 = _agent_capabilities_from_snapshot(None, "seat-1")
|
|
assert caps3 == {}
|
|
|
|
|
|
# ── 22. P1-3: Real WorkflowBuilder manager stages via WorkflowRunner ─────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_workflow_manager_stages_persisted(store):
|
|
"""P1-3: drive a REAL MAF WorkflowRunner (real WorkflowBuilder, real SQLite
|
|
repository, real adapter + seat-run store) through a structured run with a
|
|
manager + 2 seats. Only the runtime client is faked (no real LLM).
|
|
|
|
Asserts:
|
|
- manager has planner, cross_review, finalizer seat-run rows
|
|
- cross_review.turn_index == finalizer.turn_index
|
|
- cross_review.execution_stage != finalizer.execution_stage
|
|
"""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"task_text": "分析市场趋势",
|
|
"mode": "structured",
|
|
"manager_agent_id": "manager-1",
|
|
"participants": [
|
|
{"agent_id": "manager-1", "role": "manager"},
|
|
{"agent_id": "seat-a", "role": "seat"},
|
|
{"agent_id": "seat-b", "role": "seat"},
|
|
],
|
|
},
|
|
)
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
app = _make_app(store)
|
|
runner = WorkflowRunner(app, service=svc, user_id="u1")
|
|
|
|
manager_responses = iter([
|
|
"分工：seat-a 负责需求分析，seat-b 负责竞品分析",
|
|
json.dumps({
|
|
"summary": "审查通过，证据充分",
|
|
"evidence_ok": True,
|
|
"confirmed_candidates": [],
|
|
}),
|
|
"最终结论：市场看好，建议推进",
|
|
])
|
|
seat_responses = {
|
|
"seat-a": json.dumps({"summary": "需求旺盛", "findings": ["增长10%"]}),
|
|
"seat-b": json.dumps({"summary": "竞品A领先", "findings": ["份额30%"]}),
|
|
}
|
|
|
|
call_log: list[dict] = []
|
|
thread_counter = itertools.count()
|
|
created_threads: list[str] = []
|
|
latest_agent_by_thread: dict[str, str] = {}
|
|
|
|
async def fake_create_thread(principal):
|
|
thread_id = f"thread-{principal.session_id[:8]}-{next(thread_counter)}"
|
|
created_threads.append(thread_id)
|
|
return thread_id
|
|
|
|
async def fake_start_agent_run(principal, *, thread_id, agent_id, message, **kwargs):
|
|
latest_agent_by_thread[thread_id] = agent_id
|
|
call_log.append({"agent_id": agent_id, "thread_id": thread_id, "message": message})
|
|
return _FakeRunRecord(run_id=f"df-{agent_id}-{len(call_log)}")
|
|
|
|
async def fake_read_messages(principal, thread_id):
|
|
agent_id = latest_agent_by_thread[thread_id]
|
|
if agent_id == "manager-1":
|
|
content = next(manager_responses)
|
|
else:
|
|
content = seat_responses[agent_id]
|
|
return [{"type": "ai", "content": content, "usage_metadata": {"input_tokens": 5, "output_tokens": 5}}]
|
|
|
|
runner.adapter._client.create_thread = fake_create_thread
|
|
runner.adapter._client.start_agent_run = fake_start_agent_run
|
|
runner.adapter._client.read_thread_messages = fake_read_messages
|
|
runner.adapter._client.cancel_run = AsyncMock(return_value=True)
|
|
|
|
await runner.execute(s, run)
|
|
|
|
final_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert final_run["status"] == "completed"
|
|
|
|
seat_runs = await store.seat_runs.get_by_maf_run(run["id"])
|
|
mgr_rows = [r for r in seat_runs if r["agent_id"] == "manager-1"]
|
|
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}"
|
|
|
|
cr_rows = [r for r in mgr_rows if r["execution_stage"] == "cross_review"]
|
|
fin_rows = [r for r in mgr_rows if r["execution_stage"] == "finalizer"]
|
|
assert len(cr_rows) == 1 and len(fin_rows) == 1
|
|
assert cr_rows[0]["turn_index"] == fin_rows[0]["turn_index"]
|
|
assert cr_rows[0]["execution_stage"] != fin_rows[0]["execution_stage"]
|
|
|
|
assert len(created_threads) == 3
|
|
assert len(set(created_threads)) == 3
|
|
|
|
ckpt = final_run["maf_checkpoint"]
|
|
assert "最终结论：市场看好" in str(ckpt)
|
|
assert any(
|
|
r.get("agent_id") == "seat-a"
|
|
and (r.get("structured_output") or {}).get("summary") == "需求旺盛"
|
|
for r in ckpt["turn_results"]
|
|
)
|