董宏凯
1aeee86984
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema. P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events. P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback). P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks. P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
749 lines
33 KiB
Python
749 lines
33 KiB
Python
"""Stage 4 (R3-Q) runner-level tests: seat task quality loop + phase gate.
|
|
|
|
Drives the REAL workflow (QualitySeatTasksExecutor → QualityGateExecutor →
|
|
CrossReviewExecutor → ApprovalGateExecutor → FinalizerExecutor) through
|
|
WorkflowRunner.execute() with a fake adapter, proving the acceptance
|
|
criteria end-to-end against the real database:
|
|
|
|
- a good delivery is accepted and the run completes with quality_decision=accept;
|
|
- a short / evidence-free delivery is sent back, reworked up to the frozen cap,
|
|
reassigned to a backup seat once, and finally ends as a traceable
|
|
controlled_gap that is carried into the final report (never as a fact);
|
|
- an ``ask_user`` fallback pauses the run in waiting_input;
|
|
- a required task left non-terminal blocks the phase quality gate;
|
|
- a missing/invalid manager review NEVER auto-accepts: the server forces
|
|
system reworks and exhausts into a controlled gap (source=system events).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from maf_test_helpers import make_maf_service
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.deerflow_agent_adapter import SeatResponse
|
|
from app.maf_roundtable.workflow_executors import QualityGateExecutor, RoundtableState
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
Base,
|
|
MafRoundtableParticipantRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
MANAGER = "manager-1"
|
|
SEATS = ["seat-a", "seat-b", "seat-c"]
|
|
|
|
# ``@TOOL_REF@`` is replaced per-turn with ``tool:kb_search:<turn>`` — the
|
|
# trusted evidence ref created by the seat's (fake) successful tool call, so a
|
|
# good delivery cites ONLY evidence the fact ledger actually backs (P1).
|
|
GOOD_DELIVERY = json.dumps({
|
|
"summary": "结论：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，同时预留华南试点预算",
|
|
"findings": ["依据一：华东区销售数据连续三个月增长", "依据二：竞品在华东投放低于均值"],
|
|
"evidence_refs": ["@TOOL_REF@"],
|
|
"open_questions": [],
|
|
"self_check": "已覆盖全部验收标准",
|
|
})
|
|
SHORT_DELIVERY = json.dumps({"summary": "好", "findings": []})
|
|
|
|
DISPATCH_PARALLEL = {
|
|
"intent_type": "parallel_consultation",
|
|
"selected_seat_agent_ids": ["seat-a", "seat-b"],
|
|
"assignments": [
|
|
{"agent_id": "seat-a", "task_goal": "市场增长机会分析", "delivery_requirements": ["证据", "优先级"], "order": 1},
|
|
{"agent_id": "seat-b", "task_goal": "竞品定价分析", "delivery_requirements": ["证据", "建议"], "order": 2},
|
|
],
|
|
"execution_mode": "parallel",
|
|
"convergence_path": ["manager synthesis"],
|
|
"reason": "需要两位专家",
|
|
}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def q_svc(tmp_path):
|
|
"""Service with a V2 session: manager + 3 candidates, quality gate ON."""
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'q4.db'}", echo=False)
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
store = make_maf_roundtable_store(sf)
|
|
service = make_maf_service(store)
|
|
now = datetime(2026, 1, 1)
|
|
async with sf() as session:
|
|
session.add(MafRoundtableSessionRow(
|
|
id="q-sess", user_id="u1", title="q", task_text="分析市场趋势",
|
|
mode="structured", status="pending", write_policy="auto",
|
|
manager_agent_id=MANAGER, max_rounds=8, created_at=now, updated_at=now,
|
|
task_payload_json={
|
|
"maf_v2": {
|
|
"dispatch_contract_enabled": True,
|
|
"quality_gate_enabled": True,
|
|
"candidate_set": {
|
|
"manager_agent_id": MANAGER,
|
|
"candidate_seat_agent_ids": list(SEATS),
|
|
"mode": "structured",
|
|
"max_rounds": 8,
|
|
},
|
|
},
|
|
},
|
|
))
|
|
session.add(MafRoundtableRunRow(
|
|
id="q-run", session_id="q-sess", status="pending",
|
|
created_at=now, updated_at=now,
|
|
))
|
|
for i, agent_id in enumerate([MANAGER, *SEATS]):
|
|
session.add(MafRoundtableParticipantRow(
|
|
id=f"qp-{i}", session_id="q-sess", agent_id=agent_id,
|
|
role="manager" if agent_id == MANAGER else "seat",
|
|
sort_order=i, created_at=now, updated_at=now,
|
|
))
|
|
await session.commit()
|
|
yield service
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def _make_q_runner(
|
|
q_svc,
|
|
monkeypatch,
|
|
*,
|
|
deliveries: dict[str, list[str]],
|
|
review_plan: list[dict[str, dict[str, object]]],
|
|
fallback_policy: str = "reassign_then_gap",
|
|
cross_review_ok: bool = True,
|
|
emit_real: bool = False,
|
|
config_overrides: dict | None = None,
|
|
):
|
|
"""Runner with a fake adapter driven by explicit delivery/review plans.
|
|
|
|
``deliveries``: agent_id → list of JSON payloads, one per delivery attempt.
|
|
``review_plan``: one entry per quality round → {agent_id: {"decision", ...}}.
|
|
"""
|
|
from app.maf_roundtable.workflow_runner import WorkflowRunner
|
|
|
|
app = SimpleNamespace(state=SimpleNamespace(maf_roundtable_service=q_svc, checkpointer=None))
|
|
runner = WorkflowRunner(app, service=q_svc, user_id="u1")
|
|
|
|
calls: list[tuple[str, str, int]] = []
|
|
stage_messages: dict[str, list[str]] = {}
|
|
delivery_idx: dict[str, int] = {}
|
|
review_round: list[int] = [0]
|
|
|
|
cfg_values = {
|
|
"seat_timeout_seconds": 5,
|
|
"schema_repair_enabled": True,
|
|
"quality_gate_enabled": True,
|
|
"max_rework_attempts": 2,
|
|
"max_reassignments": 1,
|
|
"min_delivery_chars": 40,
|
|
"min_delivery_evidence": 1,
|
|
"default_fallback_policy": fallback_policy,
|
|
}
|
|
cfg_values.update(config_overrides or {})
|
|
fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(**cfg_values))
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)
|
|
|
|
async def _run_seat(**kwargs):
|
|
stage = kwargs.get("execution_stage", "")
|
|
agent_id = kwargs["agent_id"]
|
|
turn_index = kwargs["turn_index"]
|
|
calls.append((stage, agent_id, turn_index))
|
|
stage_messages.setdefault(stage, []).append(str(kwargs.get("message") or ""))
|
|
if stage == "manager_dispatch":
|
|
return SeatResponse(agent_id=agent_id, turn_index=0, content=json.dumps(DISPATCH_PARALLEL), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
|
|
if stage in ("expert", "rework"):
|
|
seq = deliveries.get(agent_id) or [GOOD_DELIVERY]
|
|
idx = delivery_idx.get(agent_id, 0)
|
|
delivery_idx[agent_id] = idx + 1
|
|
content = seq[min(idx, len(seq) - 1)]
|
|
# The seat ran a real (fake) tool this turn: its successful result
|
|
# becomes the trusted evidence ref the delivery may cite.
|
|
content = content.replace("@TOOL_REF@", f"tool:kb_search:{turn_index}")
|
|
return SeatResponse(
|
|
agent_id=agent_id,
|
|
turn_index=turn_index,
|
|
content=content,
|
|
tool_calls=[{"id": f"tc-{agent_id}-{turn_index}", "name": "kb_search"}],
|
|
usage={"input_tokens": 1, "output_tokens": 1},
|
|
)
|
|
if stage == "manager_review":
|
|
rnd = review_round[0]
|
|
review_round[0] = rnd + 1
|
|
plan = review_plan[rnd] if rnd < len(review_plan) else {}
|
|
decisions = [
|
|
{
|
|
"agent_id": agent,
|
|
"decision": d.get("decision", "accept"),
|
|
"reason": str(d.get("reason") or "r")[:1000],
|
|
"revision_requirements": list(d.get("requirements") or []),
|
|
}
|
|
for agent, d in plan.items()
|
|
]
|
|
return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=json.dumps({"decisions": decisions}), tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
|
|
if stage == "cross_review":
|
|
content = (
|
|
json.dumps({"summary": "ok", "evidence_ok": True, "confirmed_candidates": []})
|
|
if cross_review_ok
|
|
else json.dumps({"summary": "有缺口", "evidence_ok": False, "gaps": ["证据不足"], "confirmed_candidates": []})
|
|
)
|
|
return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=content, tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
|
|
if stage == "finalizer":
|
|
return SeatResponse(agent_id=agent_id, turn_index=turn_index, content="final report", tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
|
|
return SeatResponse(agent_id=agent_id, turn_index=turn_index, content=GOOD_DELIVERY, tool_calls=[], usage={"input_tokens": 1, "output_tokens": 1})
|
|
|
|
ensure_threads_calls: list[list[str]] = []
|
|
|
|
async def _ensure_threads(session_id, run_id, participants):
|
|
agent_ids = [p["agent_id"] for p in participants]
|
|
ensure_threads_calls.append(agent_ids)
|
|
result = {}
|
|
for participant in participants:
|
|
agent_id = participant["agent_id"]
|
|
thread_id = f"thread-{agent_id}"
|
|
await q_svc.store.participants.bind_thread(session_id, agent_id, thread_id=thread_id)
|
|
result[agent_id] = thread_id
|
|
return result
|
|
|
|
monkeypatch.setattr(runner._adapter, "run_seat_turn", AsyncMock(side_effect=_run_seat))
|
|
monkeypatch.setattr(runner._adapter, "ensure_threads", AsyncMock(side_effect=_ensure_threads))
|
|
if emit_real:
|
|
real_emit_critical = runner._adapter._emit_critical
|
|
monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock(side_effect=real_emit_critical))
|
|
else:
|
|
monkeypatch.setattr(runner._adapter, "_emit_critical", AsyncMock())
|
|
monkeypatch.setattr(runner._adapter, "_emit", AsyncMock())
|
|
runner.stage_messages = stage_messages
|
|
return runner, calls, ensure_threads_calls
|
|
|
|
|
|
async def _run(q_svc, monkeypatch, **kwargs):
|
|
runner, calls, ensure_threads_calls = _make_q_runner(q_svc, monkeypatch, **kwargs)
|
|
session = await q_svc.store.sessions.get_session("q-sess", user_id="u1")
|
|
run = await q_svc.store.runs.get_run("q-run")
|
|
await runner.execute(session, run)
|
|
final_run = await q_svc.store.runs.get_run("q-run")
|
|
final_session = await q_svc.store.sessions.get_session("q-sess", user_id="u1")
|
|
return runner, calls, ensure_threads_calls, final_run, final_session
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_accepts_good_deliveries(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
_, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [GOOD_DELIVERY], "seat-b": [GOOD_DELIVERY]},
|
|
review_plan=[{"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}}],
|
|
)
|
|
|
|
assert final_run["status"] == "completed"
|
|
assert final_session["status"] == "completed"
|
|
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
assert set(by_key) == {"01:seat-a", "02:seat-b"}
|
|
assert all(t["status"] == "accepted" for t in tasks)
|
|
assert all(t["attempt_count"] == 1 for t in tasks)
|
|
assert all(t["rework_count"] == 0 for t in tasks)
|
|
|
|
checkpoint = final_run.get("maf_checkpoint") or {}
|
|
assert checkpoint.get("seat_task_statuses", {}).get("01:seat-a", {}).get("status") == "accepted"
|
|
assert checkpoint.get("controlled_gaps") == []
|
|
|
|
result = final_session.get("final_result") or {}
|
|
assert result.get("quality_decision") == "accept"
|
|
assert result.get("controlled_gaps") == []
|
|
|
|
review_stages = [s for s, a, t in calls if s == "manager_review"]
|
|
assert len(review_stages) == 1, "one batched manager review round"
|
|
|
|
|
|
# ── 2. short answer → reworks → reassign → controlled gap ───────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_rework_reassign_controlled_gap(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
# seat-a stays short across every attempt; seat-b is always good.
|
|
# Manager keeps returning return_for_rework for seat-a (cap 2), then seat-a
|
|
# is reassigned to seat-c whose delivery also fails → controlled gap.
|
|
_, calls, ensure_threads_calls, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={
|
|
"seat-a": [SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY],
|
|
"seat-b": [GOOD_DELIVERY],
|
|
"seat-c": [SHORT_DELIVERY],
|
|
},
|
|
review_plan=[
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "太短", "requirements": ["请补充证据"]}, "seat-b": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "仍太短", "requirements": ["请展开分析"]}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
|
|
{"seat-c": {"decision": "return_for_rework", "reason": "新席位也短"}},
|
|
],
|
|
)
|
|
|
|
# Run completes via the gap path (controlled gaps do NOT block the phase).
|
|
assert final_run["status"] == "completed"
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
|
|
task_a = by_key["01:seat-a"]
|
|
assert task_a["status"] == "controlled_gap"
|
|
assert task_a["attempt_count"] == 4 # seat-a 3 + seat-c 1
|
|
assert task_a["rework_count"] == 2 # capped at max_rework_attempts
|
|
assert task_a["reassign_count"] == 1 # exactly one reassignment
|
|
assert task_a["agent_id"] == "seat-c" # reassigned to backup
|
|
assert task_a["original_agent_id"] == "seat-a"
|
|
assert task_a["gap"] is not None
|
|
assert "不得作为已验证事实" in task_a["gap"]["risk"]
|
|
|
|
task_b = by_key["02:seat-b"]
|
|
assert task_b["status"] == "accepted"
|
|
assert task_b["attempt_count"] == 1
|
|
|
|
# seat-c received a thread (backup handover) and delivered exactly once.
|
|
assert "seat-c" in [aid for call in ensure_threads_calls for aid in call]
|
|
seat_c_turns = [(s, a, t) for s, a, t in calls if a == "seat-c" and s in ("expert", "rework")]
|
|
assert len(seat_c_turns) == 1
|
|
|
|
# The quality gate passed with gaps and the final report carries them.
|
|
checkpoint = final_run.get("maf_checkpoint") or {}
|
|
gaps = checkpoint.get("controlled_gaps") or []
|
|
assert len(gaps) == 1
|
|
assert gaps[0]["task_key"] == "01:seat-a"
|
|
|
|
result = final_session.get("final_result") or {}
|
|
assert result.get("quality_decision") == "controlled_gap"
|
|
assert len(result.get("controlled_gaps") or []) == 1
|
|
assert result.get("degraded") is True
|
|
|
|
# Manager review was batched once per quality round.
|
|
review_stages = [t for s, a, t in calls if s == "manager_review"]
|
|
assert len(review_stages) == 4
|
|
|
|
|
|
# ── 3. ask_user fallback pauses in waiting_input ────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_ask_user_waits_for_input(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
_, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [SHORT_DELIVERY, SHORT_DELIVERY, SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
|
|
review_plan=[
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "仍太短"}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
|
|
],
|
|
fallback_policy="ask_user",
|
|
)
|
|
|
|
assert final_run["status"] == "waiting_input"
|
|
assert final_session["status"] == "waiting_input"
|
|
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
assert by_key["01:seat-a"]["status"] == "exhausted"
|
|
assert by_key["02:seat-b"]["status"] == "accepted"
|
|
|
|
# The finalizer must never run on the ask_user path.
|
|
finalizer_stages = [s for s, a, t in calls if s == "finalizer"]
|
|
assert finalizer_stages == []
|
|
# No finished_at for a non-terminal pause.
|
|
assert final_run.get("finished_at") is None
|
|
|
|
|
|
# ── 4. phase gate blocks on non-terminal required task ──────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quality_gate_blocks_non_terminal_required_task(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
state = RoundtableState(
|
|
session_id="q-sess", run_id="q-run", task_text="t",
|
|
mode="structured", model_name=None, reasoning_effort=None,
|
|
max_rounds=8, write_policy="auto", manager_agent_id=MANAGER,
|
|
seat_agent_ids=["seat-a"], thread_map={}, capability_snapshot=None,
|
|
seat_timeout_seconds=5, schema_repair_enabled=True,
|
|
intent_type="parallel_consultation",
|
|
dispatch_plan={"execution_mode": "parallel"},
|
|
dispatch_contract_enabled=True, quality_gate_enabled=True,
|
|
)
|
|
state.seat_task_statuses = {
|
|
"01:seat-a": {
|
|
"task_key": "01:seat-a", "status": "submitted",
|
|
"agent_id": "seat-a", "required": True,
|
|
},
|
|
}
|
|
adapter = SimpleNamespace(
|
|
_service=SimpleNamespace(store=SimpleNamespace(seat_tasks=SimpleNamespace())),
|
|
_emit_critical=AsyncMock(),
|
|
)
|
|
gate = QualityGateExecutor(adapter, state)
|
|
ctx = SimpleNamespace(send_message=AsyncMock(), yield_output=AsyncMock())
|
|
|
|
with pytest.raises(RuntimeError, match="quality gate blocked"):
|
|
await gate.handle({}, ctx)
|
|
|
|
# The blocked gate must not masquerade as passed.
|
|
assert state.status != "completed"
|
|
|
|
|
|
# ── 5. invalid/missing manager review NEVER auto-accepts ───────────────────
|
|
|
|
|
|
def _good_variant(n: int) -> str:
|
|
"""A deterministic-pass delivery whose conclusion differs per attempt, so
|
|
duplicate detection never interferes with the missing-review behaviour."""
|
|
return json.dumps({
|
|
"summary": (
|
|
f"结论{n}：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，"
|
|
f"同时预留华南试点预算，第{n}次补充细化执行路径与里程碑安排"
|
|
),
|
|
"findings": [f"依据{n}：华东区销售数据连续三个月增长", "依据二：竞品在华东投放低于均值"],
|
|
"evidence_refs": ["@TOOL_REF@"],
|
|
"open_questions": [],
|
|
"self_check": "已覆盖全部验收标准",
|
|
})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_missing_review_never_accepts(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
# The manager review returns garbage (empty decisions) every round. Even
|
|
# though every delivery passes the deterministic length/format/evidence
|
|
# checks, NO task may be accepted without a valid manager review: the
|
|
# server forces system reworks, then exhausts into controlled gaps.
|
|
variants_a = [_good_variant(i) for i in range(1, 7)]
|
|
variants_b = [_good_variant(i + 10) for i in range(1, 7)]
|
|
_, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={
|
|
"seat-a": variants_a + [_good_variant(90)],
|
|
"seat-b": variants_b + [_good_variant(91)],
|
|
"seat-c": [_good_variant(80)],
|
|
},
|
|
review_plan=[{}],
|
|
emit_real=True,
|
|
)
|
|
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
assert tasks, "seat tasks must exist"
|
|
for task in tasks:
|
|
assert task["status"] != "accepted", (
|
|
f"task {task['task_key']} was accepted without a valid manager review"
|
|
)
|
|
assert task["status"] == "controlled_gap"
|
|
assert task["rework_count"] == 2
|
|
assert task["reassign_count"] == 1
|
|
decision = task.get("review_decision") or {}
|
|
assert decision.get("source") == "system"
|
|
assert decision.get("reason") == "manager_review_missing_or_invalid"
|
|
|
|
# The run still terminates (gaps do not block the phase) — but only as a
|
|
# degraded controlled-gap result, never as a clean accept.
|
|
assert final_run["status"] == "completed"
|
|
result = final_session.get("final_result") or {}
|
|
assert result.get("quality_decision") == "controlled_gap"
|
|
assert len(result.get("controlled_gaps") or []) == len(tasks)
|
|
|
|
# The persisted review events carry source=system + the safe reason.
|
|
events = await q_svc.store.events.list_events(run_id="q-run")
|
|
review_events = [e for e in events if e["event_type"] == "manager.review.decision"]
|
|
assert review_events
|
|
system_events = [e for e in review_events if e["payload"].get("source") == "system"]
|
|
assert system_events
|
|
assert all(
|
|
e["payload"].get("reason") == "manager_review_missing_or_invalid"
|
|
for e in system_events
|
|
)
|
|
|
|
|
|
# ── 6. unaccepted deliveries never reach facts / review / final report ─────
|
|
|
|
|
|
PLAUSIBLE_GAP_DELIVERY = json.dumps({
|
|
# Fails below_min_length on purpose, but carries "plausible" findings and
|
|
# an invented evidence ref — none of it may survive into the ledger, the
|
|
# cross-review prompt or the final synthesis.
|
|
"summary": "好",
|
|
"findings": ["PLAUSIBLE_FINDING: 已核实华东区增长数据并确认竞品全面撤离"],
|
|
"evidence_refs": ["plausible-ref-001"],
|
|
"open_questions": [],
|
|
"self_check": "ok",
|
|
})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_gap_content_isolated_from_facts_and_report(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
runner, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={
|
|
"seat-a": [PLAUSIBLE_GAP_DELIVERY],
|
|
"seat-b": [GOOD_DELIVERY],
|
|
"seat-c": [PLAUSIBLE_GAP_DELIVERY],
|
|
},
|
|
review_plan=[
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "仍太短"}},
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "第三次不合格"}},
|
|
{"seat-c": {"decision": "return_for_rework", "reason": "新席位也不合格"}},
|
|
],
|
|
)
|
|
|
|
assert final_run["status"] == "completed"
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
assert by_key["01:seat-a"]["status"] == "controlled_gap"
|
|
assert by_key["02:seat-b"]["status"] == "accepted"
|
|
|
|
# 1) Fact ledger: the gapped seat's claims/refs are absent; the accepted
|
|
# seat's findings ARE present (merge happens on acceptance).
|
|
ledger = final_session.get("fact_ledger") or {}
|
|
candidate_texts = [
|
|
str(c.get("text") or "") for c in ledger.get("candidate_facts") or []
|
|
]
|
|
confirmed_texts = [
|
|
str(f.get("text") or "") for f in ledger.get("confirmed_facts") or []
|
|
]
|
|
evidence_texts = [
|
|
str(e.get("ref") or e.get("text") or "") for e in ledger.get("evidence") or []
|
|
]
|
|
assert not any("PLAUSIBLE_FINDING" in t for t in candidate_texts)
|
|
assert not any("PLAUSIBLE_FINDING" in t for t in confirmed_texts)
|
|
assert not any("plausible-ref-001" in t for t in evidence_texts)
|
|
assert any("华东区销售数据连续三个月增长" in t for t in candidate_texts)
|
|
|
|
# 2) Cross-review prompt: accepted seat only, never the gapped content.
|
|
cross_review_msgs = runner.stage_messages.get("cross_review") or []
|
|
assert cross_review_msgs
|
|
assert all("PLAUSIBLE_FINDING" not in m for m in cross_review_msgs)
|
|
assert all("plausible-ref-001" not in m for m in cross_review_msgs)
|
|
assert any("华东区销售数据连续三个月增长" in m for m in cross_review_msgs)
|
|
|
|
# 3) Final synthesis: accepted seat only + traceable gap risk section.
|
|
finalizer_msgs = runner.stage_messages.get("finalizer") or []
|
|
assert finalizer_msgs
|
|
assert all("PLAUSIBLE_FINDING" not in m for m in finalizer_msgs)
|
|
assert all("plausible-ref-001" not in m for m in finalizer_msgs)
|
|
assert any("01:seat-a" in m for m in finalizer_msgs), "gap must be traceable"
|
|
assert any("受控缺口" in m for m in finalizer_msgs)
|
|
|
|
# 4) Checkpoint: exactly one accepted result (the accepted seat).
|
|
checkpoint = final_run.get("maf_checkpoint") or {}
|
|
accepted_results = checkpoint.get("accepted_task_results") or []
|
|
assert len(accepted_results) == 1
|
|
assert accepted_results[0]["task_key"] == "02:seat-b"
|
|
assert accepted_results[0]["accepted"] is True
|
|
assert "PLAUSIBLE_FINDING" not in str(accepted_results)
|
|
|
|
# 5) Final result carries the gap as risk, never as a verified fact.
|
|
result = final_session.get("final_result") or {}
|
|
assert result.get("quality_decision") == "controlled_gap"
|
|
gaps = result.get("controlled_gaps") or []
|
|
assert len(gaps) == 1
|
|
assert gaps[0]["task_key"] == "01:seat-a"
|
|
assert "不得作为已验证事实" in gaps[0]["risk"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_events_persisted_and_projected(q_svc, monkeypatch):
|
|
"""The Stage-4 audit events persist to maf_roundtable_events."""
|
|
pytest.importorskip("agent_framework")
|
|
|
|
_, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
|
|
review_plan=[{"seat-a": {"decision": "return_for_rework", "requirements": ["补证据"]}, "seat-b": {"decision": "accept"}}],
|
|
emit_real=True,
|
|
)
|
|
|
|
events = await q_svc.store.events.list_events(run_id="q-run")
|
|
types = {e["event_type"] for e in events}
|
|
assert "seat.delivery.submitted" in types
|
|
assert "manager.review.decision" in types
|
|
assert "seat.rework.assigned" in types
|
|
assert "stage.quality_gate" in types
|
|
|
|
# The short delivery was flagged deterministically in the persisted event.
|
|
deliveries = [e for e in events if e["event_type"] == "seat.delivery.submitted"]
|
|
seat_a = [e for e in deliveries if e.get("agent_id") == "seat-a"]
|
|
assert seat_a and seat_a[0]["payload"].get("deterministic_passed") is False
|
|
|
|
|
|
# ── 7. P1: zero config values are honoured, not coerced to defaults ────────
|
|
|
|
|
|
def test_quality_settings_honour_zero_config_values(monkeypatch):
|
|
from app.maf_roundtable.workflow_runner import WorkflowRunner
|
|
|
|
fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(
|
|
quality_gate_enabled=True,
|
|
max_rework_attempts=0,
|
|
max_reassignments=0,
|
|
min_delivery_chars=0,
|
|
min_delivery_evidence=0,
|
|
default_fallback_policy="controlled_gap",
|
|
))
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)
|
|
|
|
runner = WorkflowRunner.__new__(WorkflowRunner) # no __init__ side effects
|
|
settings = runner._quality_settings()
|
|
assert settings["max_rework_attempts"] == 0
|
|
assert settings["max_reassignments"] == 0
|
|
assert settings["min_delivery_chars"] == 0
|
|
assert settings["min_delivery_evidence"] == 0
|
|
assert settings["fallback_policy"] == "controlled_gap"
|
|
|
|
|
|
def test_quality_settings_none_falls_back_to_defaults(monkeypatch):
|
|
from app.maf_roundtable.workflow_runner import WorkflowRunner
|
|
|
|
# Attributes absent entirely (getattr → None) must fall back to defaults.
|
|
fake_cfg = SimpleNamespace(maf_roundtable=SimpleNamespace(
|
|
quality_gate_enabled=True,
|
|
))
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: fake_cfg)
|
|
|
|
runner = WorkflowRunner.__new__(WorkflowRunner)
|
|
settings = runner._quality_settings()
|
|
assert settings["max_rework_attempts"] == 2
|
|
assert settings["max_reassignments"] == 1
|
|
assert settings["min_delivery_chars"] == 40
|
|
assert settings["min_delivery_evidence"] == 1
|
|
assert settings["fallback_policy"] == "reassign_then_gap"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_zero_rework_cap_no_rework_granted(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
# max_rework_attempts=0 must be honoured: a failing delivery is NOT sent
|
|
# back for rework — it exhausts straight into the fallback (controlled gap
|
|
# here, no backup reassignment budget either).
|
|
_, calls, _, final_run, final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [SHORT_DELIVERY], "seat-b": [GOOD_DELIVERY]},
|
|
review_plan=[
|
|
{"seat-a": {"decision": "return_for_rework", "reason": "太短"}, "seat-b": {"decision": "accept"}},
|
|
],
|
|
config_overrides={"max_rework_attempts": 0, "max_reassignments": 0},
|
|
)
|
|
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
task_a = by_key["01:seat-a"]
|
|
assert task_a["status"] == "controlled_gap"
|
|
assert task_a["attempt_count"] == 1, "no rework delivery may happen with a 0 cap"
|
|
assert task_a["rework_count"] == 0
|
|
assert task_a["reassign_count"] == 0
|
|
# Only the initial delivery + the (ignored) rework request — no rework turn.
|
|
expert_turns_a = [s for s, a, _t in calls if a == "seat-a" and s in ("expert", "rework")]
|
|
assert len(expert_turns_a) == 1
|
|
|
|
result = final_session.get("final_result") or {}
|
|
assert result.get("quality_decision") == "controlled_gap"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_zero_min_evidence_accepts_refless_delivery(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
# min_delivery_evidence=0 must be honoured: a conclusion-only delivery with
|
|
# no evidence refs passes the deterministic evidence gate.
|
|
no_ref_delivery = json.dumps({
|
|
"summary": "结论：该市场存在明确增长机会，建议优先进入华东区域并配套本地化运营资源投入，同时预留华南试点预算",
|
|
"findings": [],
|
|
"evidence_refs": [],
|
|
"open_questions": [],
|
|
"self_check": "ok",
|
|
})
|
|
_, _calls, _, final_run, _final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [no_ref_delivery], "seat-b": [no_ref_delivery]},
|
|
review_plan=[{"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}}],
|
|
config_overrides={"min_delivery_evidence": 0},
|
|
)
|
|
|
|
assert final_run["status"] == "completed"
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
assert all(t["status"] == "accepted" for t in tasks)
|
|
|
|
|
|
# ── 8. P1: restart recovery re-submission hits duplicate via digest ────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_quality_restart_resubmission_hits_duplicate_digest(q_svc, monkeypatch):
|
|
pytest.importorskip("agent_framework")
|
|
|
|
from app.maf_roundtable.quality_contracts import build_task_contract, conclusion_digest
|
|
|
|
good = json.loads(GOOD_DELIVERY.replace("@TOOL_REF@", "tool:kb_search:1"))
|
|
digest = conclusion_digest(good["summary"])
|
|
|
|
# Simulate the pre-restart state: seat-a's first delivery was rejected and
|
|
# its conclusion digest was persisted to the task history; then the process
|
|
# restarted. The task row (with history) survives in the DB.
|
|
contract = build_task_contract(
|
|
task_goal="市场增长机会分析", delivery_requirements=["证据", "优先级"],
|
|
max_rework_attempts=2, fallback_policy="reassign_then_gap",
|
|
)
|
|
seeded = await q_svc.store.seat_tasks.create_task(
|
|
session_id="q-sess", run_id="q-run", task_key="01:seat-a",
|
|
agent_id="seat-a", contract=contract.model_dump(), required=True,
|
|
)
|
|
await q_svc.store.seat_tasks.update_task(seeded["id"], {
|
|
"status": "rework",
|
|
"attempt_count": 1,
|
|
"rework_count": 1,
|
|
"history": [{
|
|
"attempt": 1, "agent_id": "seat-a", "round": 1,
|
|
"schema_status": "ok", "failure_codes": ["below_min_length"],
|
|
"conclusion_sha256": digest,
|
|
}],
|
|
})
|
|
|
|
# After the restart the seat re-submits the SAME conclusion. It must be
|
|
# flagged duplicate_content from the persisted digest — even though the
|
|
# in-memory prior_conclusions starts empty on a fresh process.
|
|
_, _calls, _, final_run, _final_session = await _run(
|
|
q_svc, monkeypatch,
|
|
deliveries={"seat-a": [GOOD_DELIVERY], "seat-b": [GOOD_DELIVERY]},
|
|
review_plan=[
|
|
{"seat-a": {"decision": "accept"}, "seat-b": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "accept"}},
|
|
{"seat-a": {"decision": "accept"}},
|
|
],
|
|
)
|
|
|
|
tasks = await q_svc.store.seat_tasks.list_by_run("q-run")
|
|
by_key = {t["task_key"]: t for t in tasks}
|
|
task_a = by_key["01:seat-a"]
|
|
# The re-submission was deterministically flagged as a duplicate.
|
|
check = task_a.get("deterministic_check") or {}
|
|
history_codes = [
|
|
code
|
|
for h in (task_a.get("history") or [])
|
|
for code in (h.get("failure_codes") or [])
|
|
]
|
|
assert "duplicate_content" in (check.get("failure_codes") or []) or "duplicate_content" in history_codes
|
|
# And a duplicate can never be accepted, even with the manager saying yes.
|
|
assert task_a["status"] != "accepted"
|
|
assert final_run["status"] == "completed"
|