"""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