董宏凯
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.
265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""Stage 4 (R3-Q) P1: real conversation SSE projection tests.
|
|
|
|
The Stage-4 quality events must reach the USER-VISIBLE conversation event
|
|
stream (``maf_roundtable_conversation_events``) — not just the per-run audit
|
|
table — and they must reach it safely:
|
|
|
|
1. all five Stage-4 event types are projected into the conversation table;
|
|
2. each source audit event is projected exactly ONCE (idempotent re-bridge);
|
|
3. only whitelisted fields are exposed in ``data``;
|
|
4. thread ids, run ids, internal action ids and tool parameters never leak;
|
|
5. ordinary manager / finalizer / review messages are never mislabelled as
|
|
``seat.delivery.submitted`` (only genuine seat delivery stages are).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.conversation_service import MafConversationService
|
|
from app.maf_roundtable.event_normalizer import (
|
|
make_manager_review_decision,
|
|
make_seat_delivery_submitted,
|
|
make_seat_reassigned,
|
|
make_seat_rework_assigned,
|
|
make_stage_quality_gate,
|
|
make_turn_completed,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
|
|
class FakeMafConfig:
|
|
enabled = True
|
|
default_mode = "structured"
|
|
default_model_name = "test-model"
|
|
default_reasoning_effort = None
|
|
default_manager_agent_id = "mgr-agent"
|
|
default_max_rounds = 8
|
|
default_write_policy = "confirm_high_risk"
|
|
admin_diagnostic_access = True
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def proj_env(tmp_path):
|
|
"""Store + conversation service + one conversation bound to one run."""
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'proj.db'}")
|
|
try:
|
|
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 = MafConversationService(store, maf_config=FakeMafConfig(), event_hub=None)
|
|
|
|
conv = await store.conversations.create_conversation({
|
|
"owner_user_id": "u1", "title": "projection", "task_context": None,
|
|
})
|
|
now = datetime(2026, 1, 1)
|
|
async with sf() as session:
|
|
session.add(MafRoundtableSessionRow(
|
|
id="proj-sess", user_id="u1", title="t", task_text="t",
|
|
mode="structured", status="running", write_policy="auto",
|
|
manager_agent_id="manager-1", max_rounds=8,
|
|
created_at=now, updated_at=now,
|
|
))
|
|
session.add(MafRoundtableRunRow(
|
|
id="proj-run", session_id="proj-sess", status="running",
|
|
created_at=now, updated_at=now,
|
|
))
|
|
row = await session.get(MafRoundtableConversationRow, conv["id"])
|
|
row.internal_session_id = "proj-sess"
|
|
await session.commit()
|
|
|
|
yield svc, store, conv["id"]
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def _stage4_events() -> list[tuple[str, dict]]:
|
|
"""(source_audit_event_id, event) for the five Stage-4 quality events.
|
|
|
|
Payloads are deliberately seeded with sensitive fields that must be
|
|
stripped by the projection whitelist.
|
|
"""
|
|
sensitive = {
|
|
"thread_id": "thread-secret",
|
|
"run_id": "run-secret",
|
|
"session_id": "sess-secret",
|
|
"action_id": "action-secret",
|
|
"approval_id": "approval-secret",
|
|
"target_key": "hmac-target-key",
|
|
"idempotency_key": "idem-secret",
|
|
"tool_arguments": {"url": "http://internal", "token": "abc"},
|
|
}
|
|
|
|
def _with_sensitive(event: dict) -> dict:
|
|
event = dict(event)
|
|
event["payload"] = {**event.get("payload", {}), **sensitive}
|
|
return event
|
|
|
|
return [
|
|
("audit-delivery", _with_sensitive(make_seat_delivery_submitted(
|
|
"seat-a", 1, task_key="01:seat-a", attempt=1,
|
|
deterministic_passed=False, failure_codes=["below_min_length"],
|
|
))),
|
|
("audit-review", _with_sensitive(make_manager_review_decision(
|
|
"seat-a", 1, task_key="01:seat-a", decision="rework",
|
|
reason="结论过短", source="manager", attempt=1,
|
|
))),
|
|
("audit-rework", _with_sensitive(make_seat_rework_assigned(
|
|
"seat-a", 1, task_key="01:seat-a", rework_count=1,
|
|
requirements=["请补充证据"],
|
|
))),
|
|
("audit-reassigned", _with_sensitive(make_seat_reassigned(
|
|
"seat-c", 1, task_key="01:seat-a", from_agent_id="seat-a",
|
|
reassign_count=1,
|
|
))),
|
|
("audit-gate", _with_sensitive(make_stage_quality_gate(
|
|
"manager-1", 1, decision="passed_with_gaps", total=2,
|
|
accepted=1, controlled_gap=1, blocked_required=[],
|
|
))),
|
|
]
|
|
|
|
|
|
_FORBIDDEN_KEYS = {
|
|
"thread_id", "run_id", "session_id", "action_id", "approval_id",
|
|
"target_key", "idempotency_key", "tool_arguments", "params", "arguments",
|
|
"capability_snapshot", "secret", "token",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stage4_events_projected_into_conversation_stream(proj_env):
|
|
svc, store, conv_id = proj_env
|
|
|
|
for source_id, event in _stage4_events():
|
|
written = await svc.bridge_workflow_event(
|
|
run_id="proj-run", event=event, source_audit_event_id=source_id,
|
|
)
|
|
assert written is True, f"{event['event_type']} was not projected"
|
|
|
|
events = await store.conversation_events.list_events(conv_id, after_seq=0)
|
|
types = [e["type"] for e in events]
|
|
assert "seat.delivery.submitted" in types
|
|
assert "manager.review.decision" in types
|
|
assert "seat.rework.assigned" in types
|
|
assert "seat.reassigned" in types
|
|
assert "stage.quality_gate" in types
|
|
|
|
# Whitelisted fields survive; the projection carries the quality payload.
|
|
delivery = next(e for e in events if e["type"] == "seat.delivery.submitted")
|
|
assert delivery["data"].get("task_key") == "01:seat-a"
|
|
assert delivery["data"].get("deterministic_passed") is False
|
|
assert delivery["data"].get("failure_codes") == ["below_min_length"]
|
|
review = next(e for e in events if e["type"] == "manager.review.decision")
|
|
assert review["data"].get("decision") == "rework"
|
|
assert review["data"].get("source") == "manager"
|
|
gate = next(e for e in events if e["type"] == "stage.quality_gate")
|
|
assert gate["data"].get("decision") == "passed_with_gaps"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stage4_projection_is_idempotent_per_source_event(proj_env):
|
|
svc, store, conv_id = proj_env
|
|
|
|
# Bridge every event, then bridge the SAME audit events again (retry /
|
|
# reconcile after restart). Each source event projects exactly once.
|
|
for _round in range(2):
|
|
for source_id, event in _stage4_events():
|
|
await svc.bridge_workflow_event(
|
|
run_id="proj-run", event=event, source_audit_event_id=source_id,
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv_id, after_seq=0)
|
|
projected = [
|
|
e for e in events
|
|
if e["type"] in {
|
|
"seat.delivery.submitted", "manager.review.decision",
|
|
"seat.rework.assigned", "seat.reassigned", "stage.quality_gate",
|
|
}
|
|
]
|
|
assert len(projected) == 5, f"expected exactly 5 projections, got {len(projected)}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stage4_projection_exposes_only_whitelist_fields(proj_env):
|
|
svc, store, conv_id = proj_env
|
|
|
|
for source_id, event in _stage4_events():
|
|
await svc.bridge_workflow_event(
|
|
run_id="proj-run", event=event, source_audit_event_id=source_id,
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv_id, after_seq=0)
|
|
assert events
|
|
for e in events:
|
|
data = e.get("data") or {}
|
|
leaked = _FORBIDDEN_KEYS & set(data.keys())
|
|
assert not leaked, f"{e['type']} leaks forbidden keys: {leaked}"
|
|
# Defense in depth: forbidden VALUES never appear anywhere either.
|
|
blob = str(data)
|
|
assert "thread-secret" not in blob
|
|
assert "run-secret" not in blob
|
|
assert "hmac-target-key" not in blob
|
|
assert "idem-secret" not in blob
|
|
assert "action-secret" not in blob
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manager_and_finalizer_messages_not_seat_deliveries(proj_env):
|
|
svc, store, conv_id = proj_env
|
|
|
|
# Ordinary manager-side messages: none of them may become a seat delivery.
|
|
non_delivery_stages = [
|
|
"manager_dispatch", "manager_review", "cross_review",
|
|
"cross_review_repair", "finalizer", "format_repair", "planner",
|
|
]
|
|
for i, stage in enumerate(non_delivery_stages):
|
|
await svc.bridge_workflow_event(
|
|
run_id="proj-run",
|
|
event=make_turn_completed(
|
|
"manager-1", i + 1, content=f"{stage} message",
|
|
execution_stage=stage,
|
|
),
|
|
source_audit_event_id=f"audit-msg-{stage}",
|
|
)
|
|
# A completed message WITHOUT a stage (legacy audit row) is fail-closed:
|
|
# it must not be guessed to be a seat delivery either.
|
|
await svc.bridge_workflow_event(
|
|
run_id="proj-run",
|
|
event=make_turn_completed("manager-1", 99, content="unknown stage"),
|
|
source_audit_event_id="audit-msg-nostage",
|
|
)
|
|
|
|
# A genuine seat delivery stage still projects both events.
|
|
await svc.bridge_workflow_event(
|
|
run_id="proj-run",
|
|
event=make_turn_completed(
|
|
"seat-a", 1, content='{"summary": "交付"}', execution_stage="expert",
|
|
),
|
|
source_audit_event_id="audit-msg-expert",
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv_id, after_seq=0)
|
|
deliveries = [e for e in events if e["type"] == "seat.delivery.submitted"]
|
|
assert len(deliveries) == 1, (
|
|
"only the expert seat turn may project seat.delivery.submitted, got "
|
|
f"{len(deliveries)}"
|
|
)
|
|
assert deliveries[0]["data"].get("agent_id") == "seat-a"
|
|
|
|
# All ordinary messages still appear as agent messages.
|
|
completions = [e for e in events if e["type"] == "agent.message.complete"]
|
|
assert len(completions) == len(non_delivery_stages) + 2 # + nostage + expert
|