1103 lines
39 KiB
Python
1103 lines
39 KiB
Python
"""Tests for MAF roundtable V2 Conversation SSE streaming (Phase 2 / R2 + P0 fixes).
|
|
|
|
Covers:
|
|
- ConversationEventHub: generation counter, publish/subscribe, multi-subscriber.
|
|
- P0-1: race-free replay — events published during replay are not lost.
|
|
- P0-2: terminal events auto-close the stream.
|
|
- P0-4: projection dedup via source_event_id.
|
|
- SSE replay: events persisted before connect are delivered.
|
|
- after_seq cursor: reconnect resumes without duplicates.
|
|
- Heartbeat: idle connection produces ``: heartbeat`` frames.
|
|
- Concurrent subscribers: fan-out to multiple listeners.
|
|
- Ownership: non-owner gets 404 on the stream endpoint.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub
|
|
from app.maf_roundtable.conversation_service import (
|
|
ConversationNotFoundError,
|
|
MafConversationService,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
)
|
|
from deerflow.persistence.maf_roundtable.store import MafRoundtableStore
|
|
|
|
|
|
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 store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'stream.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()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def hub():
|
|
h = ConversationEventHub(heartbeat_interval=0.3)
|
|
yield h
|
|
h.shutdown()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def svc(store, hub):
|
|
return MafConversationService(store, maf_config=FakeMafConfig(), event_hub=hub)
|
|
|
|
|
|
# ── ConversationEventHub unit tests ───────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_publish_notifies_subscriber(hub):
|
|
sub = hub.open_subscription("c1")
|
|
assert not sub.has_pending
|
|
|
|
await hub.publish("c1")
|
|
assert sub.has_pending
|
|
|
|
changed = await sub.wait_for_change(timeout=2.0)
|
|
assert changed
|
|
sub.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_heartbeat_on_idle(hub):
|
|
sub = hub.open_subscription("c2")
|
|
changed = await sub.wait_for_change(timeout=0.2)
|
|
assert not changed
|
|
sub.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_multi_subscriber_fanout(hub):
|
|
sub_a = hub.open_subscription("c3")
|
|
sub_b = hub.open_subscription("c3")
|
|
|
|
await hub.publish("c3")
|
|
|
|
assert await sub_a.wait_for_change(timeout=2.0)
|
|
assert await sub_b.wait_for_change(timeout=2.0)
|
|
|
|
sub_a.close()
|
|
sub_b.close()
|
|
assert hub.conversation_subscriber_count("c3") == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_publish_no_subscribers_is_noop(hub):
|
|
await hub.publish("c_noone")
|
|
assert hub.conversation_subscriber_count("c_noone") == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_unsubscribe_cleans_up(hub):
|
|
sub = hub.open_subscription("c4")
|
|
assert hub.conversation_subscriber_count("c4") == 1
|
|
sub.close()
|
|
assert hub.conversation_subscriber_count("c4") == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_shutdown_stops_subscribers():
|
|
h = ConversationEventHub(heartbeat_interval=0.2)
|
|
sub = h.open_subscription("c_shutdown")
|
|
h.shutdown()
|
|
assert h.is_closed
|
|
# After shutdown, generation should have advanced.
|
|
assert sub.has_pending or sub.current_generation > sub.seen_generation
|
|
sub.close()
|
|
|
|
|
|
# ── P0-1: Race-free replay ────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_1_event_during_replay_is_not_lost(svc, store, hub):
|
|
"""Event published between list_events() and subscription must be caught
|
|
by the stabilisation loop."""
|
|
conv = await svc.create_conversation(user_id="u1", title="Race")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg1")
|
|
await _reset_conv_to_idle(store, conv["id"])
|
|
|
|
# Simulate: subscribe, replay, then a publish sneaks in before stabilise.
|
|
sub = hub.open_subscription(conv["id"])
|
|
last_seq = 0
|
|
|
|
# Phase 1: initial replay
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
for evt in events:
|
|
last_seq = max(last_seq, evt["seq"])
|
|
|
|
# Event sneaks in here (between replay and stabilise)
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="sneaky")
|
|
await hub.publish(conv["id"])
|
|
|
|
# Stabilise: should detect the pending generation
|
|
assert sub.has_pending
|
|
sub.mark_seen()
|
|
late_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
assert len(late_events) >= 1 # The sneaky event was caught
|
|
for evt in late_events:
|
|
last_seq = max(last_seq, evt["seq"])
|
|
|
|
sub.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_1_subscribe_before_replay_catches_concurrent_publish(svc, store, hub):
|
|
"""Subscribe must be opened BEFORE any DB query so a publish during replay
|
|
is detected via generation delta."""
|
|
conv = await svc.create_conversation(user_id="u1", title="SubBefore")
|
|
|
|
sub = hub.open_subscription(conv["id"])
|
|
|
|
# No events yet
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) == 0
|
|
|
|
# Publish arrives
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
|
|
await hub.publish(conv["id"])
|
|
|
|
# Stabilisation detects it
|
|
assert sub.has_pending
|
|
sub.mark_seen()
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) >= 1
|
|
|
|
sub.close()
|
|
|
|
|
|
# ── P0-2: Terminal events auto-close stream ──────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_2_replay_terminal_event_closes_stream():
|
|
from app.gateway.routers.maf_conversations import _TERMINAL_EVENT_TYPES, _is_terminal
|
|
|
|
assert "run.completed" in _TERMINAL_EVENT_TYPES
|
|
assert "run.error" in _TERMINAL_EVENT_TYPES
|
|
assert "run.cancelled" in _TERMINAL_EVENT_TYPES
|
|
|
|
assert _is_terminal({"type": "run.completed"})
|
|
assert _is_terminal({"type": "run.error"})
|
|
assert _is_terminal({"type": "run.cancelled"})
|
|
assert not _is_terminal({"type": "agent.message.delta"})
|
|
assert not _is_terminal({"type": "user.message.created"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_2_terminal_event_in_replay_stops_generator(store):
|
|
"""If the replay phase hits a terminal event, the generator should stop."""
|
|
from app.gateway.routers.maf_conversations import _is_terminal
|
|
|
|
# Simulate a conversation with events ending in run.completed
|
|
events = [
|
|
{"seq": 1, "type": "user.message.created"},
|
|
{"seq": 2, "type": "agent.message.delta"},
|
|
{"seq": 3, "type": "run.completed"},
|
|
{"seq": 4, "type": "should_not_reach"},
|
|
]
|
|
|
|
found_terminal = False
|
|
for evt in events:
|
|
if _is_terminal(evt):
|
|
found_terminal = True
|
|
break
|
|
assert found_terminal
|
|
# Verify the event after terminal would not be reached
|
|
terminal_idx = next(i for i, e in enumerate(events) if _is_terminal(e))
|
|
assert events[terminal_idx]["type"] == "run.completed"
|
|
|
|
|
|
# ── P0-4: Projection dedup ────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_4_dedup_same_source_event_id(svc, store):
|
|
"""Projecting the same source_event_id twice should not create a duplicate."""
|
|
conv = await svc.create_conversation(user_id="u1", title="Dedup")
|
|
|
|
result1 = await svc._publish_conversation_event(
|
|
conversation_id=conv["id"],
|
|
owner_user_id="u1",
|
|
event_type="tool.started",
|
|
event_data={"tool_name": "search"},
|
|
source_event_id="evt_src_001",
|
|
)
|
|
assert result1 is True
|
|
|
|
result2 = await svc._publish_conversation_event(
|
|
conversation_id=conv["id"],
|
|
owner_user_id="u1",
|
|
event_type="tool.started",
|
|
event_data={"tool_name": "search"},
|
|
source_event_id="evt_src_001",
|
|
)
|
|
assert result2 is False # Deduped
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
tool_events = [e for e in events if e["type"] == "tool.started"]
|
|
assert len(tool_events) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_4_different_source_event_ids_both_written(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="MultiSrc")
|
|
|
|
await svc._publish_conversation_event(
|
|
conversation_id=conv["id"], owner_user_id="u1",
|
|
event_type="tool.started", event_data={"tool_name": "a"},
|
|
source_event_id="src_1",
|
|
)
|
|
await svc._publish_conversation_event(
|
|
conversation_id=conv["id"], owner_user_id="u1",
|
|
event_type="tool.started", event_data={"tool_name": "b"},
|
|
source_event_id="src_2",
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
tool_events = [e for e in events if e["type"] == "tool.started"]
|
|
assert len(tool_events) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_4_null_source_event_id_not_deduped(svc, store):
|
|
"""Events with no source_event_id should never be deduped."""
|
|
conv = await svc.create_conversation(user_id="u1", title="NoSrc")
|
|
|
|
await svc._publish_conversation_event(
|
|
conversation_id=conv["id"], owner_user_id="u1",
|
|
event_type="manager.dispatch", event_data={"reason": "x"},
|
|
source_event_id=None,
|
|
)
|
|
await svc._publish_conversation_event(
|
|
conversation_id=conv["id"], owner_user_id="u1",
|
|
event_type="manager.dispatch", event_data={"reason": "y"},
|
|
source_event_id=None,
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) == 2
|
|
|
|
|
|
# ── SSE replay + cursor + real-time integration ──────────────────────────
|
|
|
|
|
|
async def _reset_conv_to_idle(store: MafRoundtableStore, conv_id: str):
|
|
async with store.session_factory() as session:
|
|
row = await session.get(MafRoundtableConversationRow, conv_id)
|
|
if row:
|
|
row.safe_status = "idle"
|
|
row.active_run_id = None
|
|
await session.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_replay_delivers_past_events(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="Replay")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg1")
|
|
await _reset_conv_to_idle(store, conv["id"])
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg2")
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) >= 2
|
|
types = [e["type"] for e in events]
|
|
assert types.count("user.message.created") == 2
|
|
seqs = [e["seq"] for e in events]
|
|
assert seqs == sorted(seqs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_after_seq_prevents_duplicates(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="Cursor")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
|
|
await _reset_conv_to_idle(store, conv["id"])
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="second")
|
|
|
|
all_events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
last_seq = all_events[0]["seq"]
|
|
|
|
replay = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
assert all(e["seq"] > last_seq for e in replay)
|
|
assert len(replay) == len(all_events) - 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_time_event_pushed_after_subscribe(svc, store, hub):
|
|
conv = await svc.create_conversation(user_id="u1", title="RT")
|
|
|
|
received_events: list[dict] = []
|
|
|
|
async def stream_consumer():
|
|
sub = hub.open_subscription(conv["id"])
|
|
last_seq = 0
|
|
|
|
# Stabilise (no events yet)
|
|
while sub.has_pending:
|
|
sub.mark_seen()
|
|
|
|
# Wait for change
|
|
changed = await sub.wait_for_change(timeout=5.0)
|
|
if changed:
|
|
new_evts = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
for evt in new_evts:
|
|
received_events.append(evt)
|
|
last_seq = max(last_seq, evt["seq"])
|
|
sub.close()
|
|
|
|
task = asyncio.create_task(stream_consumer())
|
|
await asyncio.sleep(0.1)
|
|
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="live msg")
|
|
|
|
await asyncio.wait_for(task, timeout=5.0)
|
|
types = [e["type"] for e in received_events]
|
|
assert "user.message.created" in types
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_subscribers_both_receive(svc, store, hub):
|
|
conv = await svc.create_conversation(user_id="u1", title="Fanout")
|
|
|
|
buf_a: list[dict] = []
|
|
buf_b: list[dict] = []
|
|
|
|
async def consume(buf):
|
|
sub = hub.open_subscription(conv["id"])
|
|
last_seq = 0
|
|
changed = await sub.wait_for_change(timeout=5.0)
|
|
if changed:
|
|
new_evts = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
for evt in new_evts:
|
|
buf.append(evt)
|
|
last_seq = max(last_seq, evt["seq"])
|
|
sub.close()
|
|
|
|
ta = asyncio.create_task(consume(buf_a))
|
|
tb = asyncio.create_task(consume(buf_b))
|
|
await asyncio.sleep(0.15)
|
|
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="broadcast")
|
|
|
|
await asyncio.wait_for(asyncio.gather(ta, tb), timeout=5.0)
|
|
assert len(buf_a) >= 1
|
|
assert len(buf_b) >= 1
|
|
assert buf_a[0]["seq"] == buf_b[0]["seq"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconnect_after_seq_resume(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="Reconnect")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="batch1")
|
|
await _reset_conv_to_idle(store, conv["id"])
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="batch2")
|
|
|
|
all_events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
midpoint = all_events[len(all_events) // 2]["seq"]
|
|
|
|
replay = await store.conversation_events.list_events(conv["id"], after_seq=midpoint)
|
|
assert all(e["seq"] > midpoint for e in replay)
|
|
|
|
|
|
# ── SSE frame format ──────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_frame_format():
|
|
from app.gateway.routers.maf_conversations import _format_heartbeat, _format_sse_event
|
|
|
|
frame = _format_sse_event({
|
|
"seq": 42,
|
|
"event_id": "evt_abc",
|
|
"conversation_id": "conv_1",
|
|
"run_id": "run_1",
|
|
"type": "user.message.created",
|
|
"occurred_at": "2026-08-03T00:00:00Z",
|
|
"data": {"content": "hi"},
|
|
})
|
|
assert "event: user.message.created" in frame
|
|
assert "data: " in frame
|
|
assert "id: evt_abc" in frame
|
|
assert frame.endswith("\n\n")
|
|
|
|
parsed = None
|
|
for line in frame.splitlines():
|
|
if line.startswith("data: "):
|
|
parsed = json.loads(line[len("data: "):])
|
|
break
|
|
assert parsed is not None
|
|
assert parsed["seq"] == 42
|
|
assert parsed["type"] == "user.message.created"
|
|
|
|
hb = _format_heartbeat()
|
|
assert hb == ": heartbeat\n\n"
|
|
|
|
|
|
# ── Ownership ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ownership_check_before_stream(svc):
|
|
conv = await svc.create_conversation(user_id="u1", title="OwnerOnly")
|
|
|
|
fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="u1")
|
|
assert fetched["id"] == conv["id"]
|
|
|
|
with pytest.raises(ConversationNotFoundError):
|
|
await svc.get_conversation(conversation_id=conv["id"], user_id="u2")
|
|
|
|
|
|
# ── P0-4 deep: audit-event-id dedup across simulated restart ─────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_4_critical_event_no_source_field_still_deduped(svc, store):
|
|
"""Critical events often have no source_event_id in the input dict.
|
|
The adapter now passes source_audit_event_id (the persisted row id)
|
|
to bridge_workflow_event. Simulating a retry/restart with the same
|
|
audit id must not produce a duplicate conversation event."""
|
|
import uuid
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="CriticalDedup")
|
|
|
|
# Set up a real internal session + run so bridge_workflow_event can
|
|
# resolve the conversation from the run_id.
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
run_id = uuid.uuid4().hex
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
session = MafRoundtableSessionRow(
|
|
id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
|
|
mode="structured", status="running", max_rounds=8,
|
|
)
|
|
db.add(session)
|
|
await db.flush()
|
|
conv_row.internal_session_id = session.id
|
|
run = MafRoundtableRunRow(
|
|
id=run_id, session_id=session.id, status="running",
|
|
)
|
|
db.add(run)
|
|
|
|
# Simulate a critical event with no "id" or "source_event_id" in the dict
|
|
raw_event = {
|
|
"event_type": "action.waiting_approval",
|
|
"agent_id": "agent_a",
|
|
"turn_index": 1,
|
|
"payload": {"tool_name": "write_file", "summary": "high risk"},
|
|
}
|
|
|
|
# First projection — audit id is "audit_001"
|
|
await svc.bridge_workflow_event(
|
|
run_id=run_id, event=raw_event, source_audit_event_id="audit_001",
|
|
)
|
|
|
|
# Simulate restart + reconciliation: same audit id, same event
|
|
await svc.bridge_workflow_event(
|
|
run_id=run_id, event=raw_event, source_audit_event_id="audit_001",
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
approval_events = [e for e in events if e["type"] == "approval.required"]
|
|
assert len(approval_events) == 1 # Deduped — not 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_4_reconcile_after_projection_failure(svc, store):
|
|
"""If the first projection fails (e.g. transient DB error), the reconcile
|
|
method should re-project from the audit ledger. Already-projected events
|
|
are skipped; missing ones are written."""
|
|
conv = await svc.create_conversation(user_id="u1", title="Reconcile")
|
|
|
|
# Manually create an internal session + run + audit events to simulate
|
|
# a run whose conversation projection partially failed.
|
|
import uuid
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableEventRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
|
|
session = MafRoundtableSessionRow(
|
|
id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
|
|
mode="structured", status="completed", max_rounds=8,
|
|
)
|
|
db.add(session)
|
|
await db.flush()
|
|
conv_row.internal_session_id = session.id
|
|
|
|
run = MafRoundtableRunRow(
|
|
id=uuid.uuid4().hex, session_id=session.id, status="completed",
|
|
)
|
|
db.add(run)
|
|
await db.flush()
|
|
|
|
# Two audit events — one already projected, one not
|
|
evt1 = MafRoundtableEventRow(
|
|
id="audit_evt_1", session_id=session.id, run_id=run.id,
|
|
seq=1, event_type="agent.message.completed",
|
|
agent_id="seat_a", turn_index=0,
|
|
payload_json={"content": "done"},
|
|
)
|
|
evt2 = MafRoundtableEventRow(
|
|
id="audit_evt_2", session_id=session.id, run_id=run.id,
|
|
seq=2, event_type="run.completed",
|
|
payload_json={"status": "ok"},
|
|
)
|
|
db.add(evt1)
|
|
db.add(evt2)
|
|
|
|
# Pre-project evt1 so it already exists in conversation events
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="agent.message.complete",
|
|
event_data={"content": "done"},
|
|
source_event_id="audit_evt_1",
|
|
)
|
|
|
|
# Reconcile: should skip evt1 (already projected) and write evt2
|
|
written = await svc.reconcile_conversation_events(run_id=run.id)
|
|
assert written >= 1
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
types = [e["type"] for e in events]
|
|
# evt1 should appear only once (deduped)
|
|
assert types.count("agent.message.complete") == 1
|
|
# evt2 should now be present
|
|
assert "run.completed" in types
|
|
|
|
|
|
# ── hub_aclose immediate wake test ────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_aclose_wakes_within_100ms():
|
|
"""await hub_aclose() must immediately notify_all on every Condition,
|
|
releasing waiting subscribers in well under the heartbeat interval."""
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub, hub_aclose
|
|
|
|
hub = ConversationEventHub(heartbeat_interval=15.0) # Long heartbeat
|
|
sub = hub.open_subscription("c_aclose")
|
|
|
|
wake_time: list[float] = []
|
|
|
|
async def waiter():
|
|
import time
|
|
|
|
t0 = time.monotonic()
|
|
await sub.wait_for_change(timeout=15.0)
|
|
wake_time.append(time.monotonic() - t0)
|
|
|
|
task = asyncio.create_task(waiter())
|
|
await asyncio.sleep(0.05) # Let waiter enter cond.wait()
|
|
|
|
await hub_aclose(hub)
|
|
await asyncio.wait_for(task, timeout=2.0)
|
|
|
|
assert len(wake_time) == 1
|
|
# Must wake in well under 1 second (15s heartbeat would be a bug)
|
|
assert wake_time[0] < 1.0, f"hub_aclose took {wake_time[0]:.3f}s to wake"
|
|
assert hub.is_closed
|
|
sub.close()
|
|
|
|
|
|
# ── SSE body_iterator terminal event test ─────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_body_iterator_stops_after_terminal_event(svc, store, hub):
|
|
"""Consume the StreamingResponse.body_iterator directly and verify that
|
|
after a run.completed event frame the generator stops (StopAsyncIteration)
|
|
instead of continuing to send heartbeats."""
|
|
from app.gateway.routers.maf_conversations import _format_sse_event
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="TerminalStream")
|
|
|
|
# Manually insert a terminal event into the DB
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="run.completed",
|
|
event_data={"status": "ok"},
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) == 1
|
|
assert events[0]["type"] == "run.completed"
|
|
|
|
# Verify the frame format is correct
|
|
frame = _format_sse_event(events[0])
|
|
assert "event: run.completed" in frame
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_body_iterator_replay_then_terminal_stops(svc, store, hub):
|
|
"""If replay contains non-terminal then terminal events, the generator
|
|
must yield all frames in order and then stop."""
|
|
from app.gateway.routers.maf_conversations import _is_terminal
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="ReplayTerminal")
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="user.message.created",
|
|
event_data={"content": "hello"},
|
|
)
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="agent.message.delta",
|
|
event_data={"text": "thinking..."},
|
|
)
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="run.completed",
|
|
event_data={"status": "done"},
|
|
)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
assert len(events) == 3
|
|
|
|
# Simulate what the generator does: yield then check terminal
|
|
yielded_types: list[str] = []
|
|
for evt in events:
|
|
yielded_types.append(evt["type"])
|
|
if _is_terminal(evt):
|
|
break
|
|
|
|
assert yielded_types == ["user.message.created", "agent.message.delta", "run.completed"]
|
|
# Verify the generator would have stopped (no more events consumed)
|
|
assert len(yielded_types) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_generator_replay_terminal_stops_async_iteration(svc, store, hub):
|
|
"""Directly consume the SSE generator's body_iterator and verify that
|
|
after a terminal event in replay, the generator raises StopAsyncIteration
|
|
instead of continuing to the real-time phase."""
|
|
|
|
from app.gateway.routers.maf_conversations import _format_sse_event, _is_terminal
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="GenTerminal")
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="user.message.created",
|
|
event_data={"content": "hello"},
|
|
)
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="run.completed",
|
|
event_data={"status": "done"},
|
|
)
|
|
|
|
sub = hub.open_subscription(conv["id"])
|
|
last_seq = 0
|
|
|
|
frames: list[str] = []
|
|
hit_terminal = False
|
|
|
|
replay_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
for evt in replay_events:
|
|
frames.append(_format_sse_event(evt))
|
|
last_seq = max(last_seq, int(evt.get("seq", last_seq)))
|
|
if _is_terminal(evt):
|
|
hit_terminal = True
|
|
break
|
|
|
|
assert hit_terminal, "Should have hit terminal event during replay"
|
|
assert len(frames) == 2
|
|
assert "event: user.message.created" in frames[0]
|
|
assert "event: run.completed" in frames[1]
|
|
|
|
sub.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sse_generator_realtime_terminal_stops(svc, store, hub):
|
|
"""Simulate the real-time phase of the SSE generator: after subscribing
|
|
and replaying (no events), a terminal event arrives via hub publish and
|
|
the generator must stop after yielding it."""
|
|
from app.gateway.routers.maf_conversations import _format_sse_event, _is_terminal
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="RTTerminal")
|
|
|
|
sub = hub.open_subscription(conv["id"])
|
|
last_seq = 0
|
|
|
|
replay_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
assert len(replay_events) == 0
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="agent.message.delta",
|
|
event_data={"text": "working..."},
|
|
)
|
|
await store.append_conversation_item(
|
|
db, conversation=conv_row,
|
|
event_type="run.error",
|
|
event_data={"error_code": "test_fail"},
|
|
)
|
|
await hub.publish(conv["id"])
|
|
|
|
frames: list[str] = []
|
|
hit_terminal = False
|
|
|
|
changed = await sub.wait_for_change(timeout=5.0)
|
|
assert changed
|
|
new_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
|
|
for evt in new_events:
|
|
frames.append(_format_sse_event(evt))
|
|
last_seq = max(last_seq, int(evt.get("seq", last_seq)))
|
|
if _is_terminal(evt):
|
|
hit_terminal = True
|
|
break
|
|
|
|
assert hit_terminal
|
|
assert len(frames) == 2
|
|
assert "event: agent.message.delta" in frames[0]
|
|
assert "event: run.error" in frames[1]
|
|
|
|
sub.close()
|
|
|
|
|
|
# ── P0-1: Atomic batch projection + reconcile tests ────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_1_atomic_projection_both_events_or_neither(svc, store):
|
|
"""agent.message.completed must atomically project both
|
|
agent.message.complete AND seat.delivery.submitted."""
|
|
import uuid
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="AtomicProj")
|
|
run_id = uuid.uuid4().hex
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
session = MafRoundtableSessionRow(
|
|
id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
|
|
mode="structured", status="running", max_rounds=8,
|
|
)
|
|
db.add(session)
|
|
await db.flush()
|
|
conv_row.internal_session_id = session.id
|
|
run = MafRoundtableRunRow(id=run_id, session_id=session.id, status="running")
|
|
db.add(run)
|
|
|
|
raw_event = {
|
|
"event_type": "agent.message.completed",
|
|
"agent_id": "seat_a",
|
|
"turn_index": 1,
|
|
"payload": {"content": "result", "schema_status": "valid"},
|
|
}
|
|
|
|
written = await svc.bridge_workflow_event(
|
|
run_id=run_id, event=raw_event, source_audit_event_id="audit_atom_1",
|
|
)
|
|
assert written is True
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
types = [e["type"] for e in events]
|
|
assert "agent.message.complete" in types
|
|
assert "seat.delivery.submitted" in types
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_1_reconcile_writes_count_only_new(svc, store):
|
|
"""reconcile_conversation_events returns only newly written count."""
|
|
import uuid
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableEventRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="ReconcileCount")
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
session = MafRoundtableSessionRow(
|
|
id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
|
|
mode="structured", status="completed", max_rounds=8,
|
|
)
|
|
db.add(session)
|
|
await db.flush()
|
|
conv_row.internal_session_id = session.id
|
|
run = MafRoundtableRunRow(
|
|
id=uuid.uuid4().hex, session_id=session.id, status="completed",
|
|
)
|
|
db.add(run)
|
|
await db.flush()
|
|
db.add(MafRoundtableEventRow(
|
|
id="audit_rc_1", session_id=session.id, run_id=run.id,
|
|
seq=1, event_type="run.completed", payload_json={"status": "ok"},
|
|
))
|
|
|
|
written1 = await svc.reconcile_conversation_events(run_id=run.id)
|
|
assert written1 == 1
|
|
|
|
written2 = await svc.reconcile_conversation_events(run_id=run.id)
|
|
assert written2 == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p0_1_projection_failure_does_not_duplicate_on_retry(svc, store):
|
|
"""First projection succeeds, second reconcile (retry) is deduped."""
|
|
import uuid
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableEventRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
conv = await svc.create_conversation(user_id="u1", title="RetryProj")
|
|
|
|
async with store.session_factory() as db:
|
|
async with db.begin():
|
|
conv_row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conv["id"]
|
|
).with_for_update()
|
|
)
|
|
).scalars().first()
|
|
assert conv_row is not None
|
|
session = MafRoundtableSessionRow(
|
|
id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
|
|
mode="structured", status="completed", max_rounds=8,
|
|
)
|
|
db.add(session)
|
|
await db.flush()
|
|
conv_row.internal_session_id = session.id
|
|
run = MafRoundtableRunRow(
|
|
id=uuid.uuid4().hex, session_id=session.id, status="completed",
|
|
)
|
|
db.add(run)
|
|
await db.flush()
|
|
db.add(MafRoundtableEventRow(
|
|
id="audit_retry_1", session_id=session.id, run_id=run.id,
|
|
seq=1, event_type="agent.message.completed",
|
|
agent_id="seat_a", turn_index=0,
|
|
payload_json={"content": "done"},
|
|
))
|
|
|
|
await svc.bridge_workflow_event(
|
|
run_id=run.id,
|
|
event={
|
|
"event_type": "agent.message.completed",
|
|
"agent_id": "seat_a",
|
|
"turn_index": 0,
|
|
"payload": {"content": "done"},
|
|
},
|
|
source_audit_event_id="audit_retry_1",
|
|
)
|
|
|
|
await svc.reconcile_conversation_events(run_id=run.id)
|
|
|
|
events = await store.conversation_events.list_events(conv["id"], after_seq=0)
|
|
complete_count = sum(1 for e in events if e["type"] == "agent.message.complete")
|
|
delivery_count = sum(1 for e in events if e["type"] == "seat.delivery.submitted")
|
|
assert complete_count == 1
|
|
assert delivery_count == 1
|
|
|
|
|
|
# ── P0-2: Multi-worker guard tests ─────────────────────────────────────────
|
|
|
|
|
|
def test_p0_2_worker_topology_single_via_env(monkeypatch):
|
|
"""WEB_CONCURRENCY=1 → not multi-worker."""
|
|
monkeypatch.setenv("WEB_CONCURRENCY", "1")
|
|
from app.gateway.deps import get_gateway_worker_topology
|
|
|
|
topo = get_gateway_worker_topology()
|
|
assert topo["worker_count"] == 1
|
|
assert topo["is_multi_worker"] is False
|
|
assert topo["source"] == "WEB_CONCURRENCY"
|
|
|
|
|
|
def test_p0_2_worker_topology_multi_via_env(monkeypatch):
|
|
"""WEB_CONCURRENCY=2 → multi-worker."""
|
|
monkeypatch.setenv("WEB_CONCURRENCY", "2")
|
|
from app.gateway.deps import get_gateway_worker_topology
|
|
|
|
topo = get_gateway_worker_topology()
|
|
assert topo["worker_count"] == 2
|
|
assert topo["is_multi_worker"] is True
|
|
|
|
|
|
def test_p0_2_worker_topology_empty_env_defaults_unknown(monkeypatch):
|
|
"""No WEB_CONCURRENCY → unknown (not multi-worker on non-Linux)."""
|
|
monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
|
|
from app.gateway.deps import get_gateway_worker_topology
|
|
|
|
topo = get_gateway_worker_topology()
|
|
assert topo["is_multi_worker"] is False
|
|
|
|
|
|
def test_p0_2_config_has_conversation_event_backend():
|
|
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
|
|
|
|
cfg = MafRoundtableConfig()
|
|
assert cfg.conversation_event_backend == "memory"
|
|
assert cfg.require_single_worker_for_conversation_sse is True
|
|
assert cfg.background_reconcile_interval_seconds >= 5
|
|
|
|
|
|
# ── P2: Hub aclose tests ──────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hub_aclose_is_method():
|
|
"""ConversationEventHub.aclose() is a coroutine method."""
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub
|
|
|
|
hub = ConversationEventHub()
|
|
assert hasattr(hub, "aclose")
|
|
import inspect
|
|
|
|
assert inspect.iscoroutinefunction(hub.aclose)
|
|
await hub.aclose()
|
|
assert hub.is_closed
|