586 lines
25 KiB
Python
586 lines
25 KiB
Python
"""Tests for the MAF DeerFlow Agent Adapter (Phase 2 rework).
|
|
|
|
Tests: event normalization, context building, sanitization, cancellation,
|
|
audit event reliability, seat run persistence, and user isolation.
|
|
Uses in-memory SQLite + mocks for the DeerFlow runtime (no real model calls).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.context_builder import build_manager_synthesis_input, build_seat_input
|
|
from app.maf_roundtable.deerflow_agent_adapter import (
|
|
AuditEventError,
|
|
DeerFlowAgentAdapter,
|
|
SeatResponse,
|
|
)
|
|
from app.maf_roundtable.event_normalizer import (
|
|
make_run_completed,
|
|
make_turn_completed,
|
|
make_turn_started,
|
|
normalize_stream_event,
|
|
)
|
|
from app.maf_roundtable.runtime_client import DeerFlowRuntimeClient, ExecutionPrincipal
|
|
from app.maf_roundtable.sanitize import sanitize, sanitize_payload
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import Base
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store():
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
s = make_maf_roundtable_store(sf)
|
|
assert s is not None
|
|
yield s
|
|
await engine.dispose()
|
|
|
|
|
|
def _svc(store, enabled: bool = True):
|
|
from maf_test_helpers import make_maf_service
|
|
|
|
return make_maf_service(store, enabled=enabled)
|
|
|
|
|
|
def _mock_client() -> DeerFlowRuntimeClient:
|
|
app = MagicMock()
|
|
app.state.checkpointer = AsyncMock()
|
|
app.state.thread_store = AsyncMock()
|
|
app.state.stream_bridge = None
|
|
app.state.run_manager = AsyncMock()
|
|
return DeerFlowRuntimeClient(app)
|
|
|
|
|
|
# ── event_normalizer tests ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestEventNormalizer:
|
|
def test_metadata_event(self):
|
|
events = normalize_stream_event("metadata", {"run_id": "r1", "thread_id": "t1"}, agent_id="a1", turn_index=0)
|
|
assert len(events) == 1
|
|
assert events[0]["event_type"] == "run.started"
|
|
|
|
def test_error_event(self):
|
|
events = normalize_stream_event("error", {"message": "boom", "name": "TimeoutError"}, agent_id="a1", turn_index=1)
|
|
assert len(events) == 1
|
|
assert events[0]["event_type"] == "run.failed"
|
|
|
|
def test_messages_ai_with_tool_calls(self):
|
|
data = ["msg-1", {"type": "ai", "content": "", "tool_calls": [{"name": "search", "id": "tc1", "args": {"q": "test"}}]}]
|
|
events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
|
|
assert any(e["event_type"] == "tool.requested" for e in events)
|
|
|
|
def test_messages_tool_result(self):
|
|
data = ["msg-1", {"type": "tool", "name": "search", "tool_call_id": "tc1", "content": "results...", "status": "completed"}]
|
|
events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
|
|
assert events[0]["event_type"] == "tool.completed"
|
|
|
|
def test_no_skill_guessing_from_text(self):
|
|
data = {"agent_node": {"messages": [{"type": "ai", "content": "I will use the skill to analyze this"}]}}
|
|
events = normalize_stream_event("updates", data, agent_id="a1", turn_index=1)
|
|
assert not any("skill" in e["event_type"] for e in events)
|
|
|
|
def test_source_event_id_for_dedup(self):
|
|
data = ["msg-1", {"type": "tool", "name": "search", "tool_call_id": "tc1", "content": "ok", "status": "completed"}]
|
|
events = normalize_stream_event("messages", data, agent_id="a1", turn_index=1)
|
|
assert events[0].get("source_event_id") == "tool:tc1"
|
|
|
|
def test_unknown_event_ignored(self):
|
|
events = normalize_stream_event("checkpoints", {"foo": "bar"}, agent_id="a1", turn_index=0)
|
|
assert events == []
|
|
|
|
|
|
# ── sanitization tests ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSanitization:
|
|
def test_sensitive_keys_redacted(self):
|
|
data = {"token": "abc123", "name": "safe", "nested": {"password": "secret", "ok": "yes"}}
|
|
result = sanitize(data)
|
|
assert result["token"] == "***REDACTED***"
|
|
assert result["name"] == "safe"
|
|
assert result["nested"]["password"] == "***REDACTED***"
|
|
assert result["nested"]["ok"] == "yes"
|
|
|
|
def test_bearer_token_redacted(self):
|
|
data = {"header": "Bearer eyJhbGciOiJIUzI1NiJ9.payload.sig"}
|
|
result = sanitize(data)
|
|
assert result["header"] == "***REDACTED***"
|
|
|
|
def test_api_key_pattern_redacted(self):
|
|
data = {"key": "sk-abcdefghijklmnopqrstuvwxyz123456"}
|
|
result = sanitize(data)
|
|
assert result["key"] == "***REDACTED***"
|
|
|
|
def test_normal_values_untouched(self):
|
|
data = {"message": "hello world", "count": 42, "items": ["a", "b"]}
|
|
result = sanitize(data)
|
|
assert result == data
|
|
|
|
def test_sanitize_payload_none(self):
|
|
assert sanitize_payload(None) is None
|
|
|
|
def test_event_payload_sanitized(self):
|
|
events = normalize_stream_event(
|
|
"messages",
|
|
["m1", {"type": "ai", "content": "", "tool_calls": [{"name": "api_call", "id": "tc1", "args": {"api_key": "secret123", "url": "http://x"}}]}],
|
|
agent_id="a1",
|
|
turn_index=1,
|
|
)
|
|
tool_ev = next(e for e in events if e["event_type"] == "tool.requested")
|
|
assert tool_ev["payload"]["args"]["api_key"] == "***REDACTED***"
|
|
assert tool_ev["payload"]["args"]["url"] == "http://x"
|
|
|
|
|
|
# ── context_builder tests ─────────────────────────────────────────────────────
|
|
|
|
|
|
class TestContextBuilder:
|
|
def test_strict_truncation(self):
|
|
result = build_seat_input(
|
|
task_text="x" * 20000,
|
|
stage_instruction="",
|
|
role_description="",
|
|
prior_turns=[],
|
|
max_context_chars=100,
|
|
)
|
|
assert len(result) <= 100
|
|
|
|
def test_no_overflow_after_truncation(self):
|
|
result = build_seat_input(
|
|
task_text="task",
|
|
stage_instruction="stage",
|
|
role_description="role",
|
|
prior_turns=[{"agent_id": "a", "turn_index": 1, "summary": "y" * 5000}],
|
|
fact_ledger={"confirmed_facts": ["f" * 5000]},
|
|
max_context_chars=200,
|
|
)
|
|
assert len(result) <= 200
|
|
|
|
def test_fact_ledger_sanitized(self):
|
|
ledger = {"confirmed_facts": ["server uses token=abc123"]}
|
|
result = build_seat_input(
|
|
task_text="check",
|
|
stage_instruction="",
|
|
role_description="",
|
|
prior_turns=[],
|
|
fact_ledger=ledger,
|
|
)
|
|
assert "abc123" not in result or "token" not in result.lower()
|
|
|
|
def test_manager_synthesis(self):
|
|
result = build_manager_synthesis_input(
|
|
task_text="evaluate",
|
|
seat_results=[{"agent_id": "risk", "content": "high risk"}],
|
|
)
|
|
assert "risk" in result
|
|
assert "建议动作" in result
|
|
|
|
|
|
# ── ExecutionPrincipal tests ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestExecutionPrincipal:
|
|
def test_requires_user_id(self):
|
|
with pytest.raises(ValueError):
|
|
ExecutionPrincipal(user_id="", session_id="s1", run_id="r1")
|
|
|
|
def test_valid_principal(self):
|
|
p = ExecutionPrincipal(user_id="u1", session_id="s1", run_id="r1")
|
|
assert p.user_id == "u1"
|
|
|
|
def test_satisfies_current_user_protocol(self):
|
|
from app.maf_roundtable.runtime_client import _PrincipalUser
|
|
from deerflow.runtime.user_context import CurrentUser
|
|
|
|
p = ExecutionPrincipal(user_id="test-user", session_id="s", run_id="r")
|
|
user = _PrincipalUser(p)
|
|
assert isinstance(user, CurrentUser)
|
|
assert user.id == "test-user"
|
|
|
|
|
|
# ── adapter unit tests ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestDeerFlowAgentAdapter:
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_threads_creates_and_binds(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "test", "participants": [{"agent_id": "a1"}, {"agent_id": "a2"}]},
|
|
)
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
client = _mock_client()
|
|
client._app.state.checkpointer.aput = AsyncMock()
|
|
client._app.state.thread_store.create = AsyncMock()
|
|
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
thread_map = await adapter.ensure_threads(session["id"], run["id"], session["participants"])
|
|
|
|
assert len(thread_map) == 2
|
|
assert thread_map["a1"] != thread_map["a2"]
|
|
|
|
updated = await svc.get_session(session_id=session["id"], user_id="u1")
|
|
for p in updated["participants"]:
|
|
assert p["thread_id"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_threads_reuses_existing(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "test", "participants": [{"agent_id": "a1", "thread_id": "existing-t1"}]},
|
|
)
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
thread_map = await adapter.ensure_threads(session["id"], run["id"], session["participants"])
|
|
assert thread_map["a1"] == "existing-t1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_cascades_via_seat_runs(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
seat_run_id = active[0]["id"]
|
|
await svc.store.seat_runs.mark_running(seat_run_id, deerflow_run_id="df-run-1")
|
|
|
|
client = _mock_client()
|
|
client._app.state.run_manager.cancel = AsyncMock()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
result = await adapter.cancel(run["id"])
|
|
|
|
# Structured cancel result (Task 4): all_cancelled when every active
|
|
# sub-run was confirmed cancelled.
|
|
assert result["all_cancelled"] is True
|
|
assert result["cancelled_run_ids"] == [seat_run_id]
|
|
assert result["failed_run_ids"] == []
|
|
client._app.state.run_manager.cancel.assert_called_once_with("df-run-1", action="interrupt")
|
|
remaining = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
assert len(remaining) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_critical_event_failure_raises(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
original_append = svc.append_event
|
|
async def _failing_append(**kwargs):
|
|
raise RuntimeError("DB connection lost")
|
|
svc.append_event = _failing_append
|
|
|
|
with pytest.raises(AuditEventError):
|
|
await adapter._emit_critical(run["id"], make_turn_started("a1", 0))
|
|
|
|
svc.append_event = original_append
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_noncritical_event_degrades_gracefully(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
async def _failing_append(**kwargs):
|
|
raise RuntimeError("DB connection lost")
|
|
svc.append_event = _failing_append
|
|
|
|
await adapter._emit(run["id"], {"event_type": "agent.message.delta", "agent_id": "a1", "turn_index": 0, "payload": {"text": "hi"}})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seat_response_dataclass(self):
|
|
r = SeatResponse(agent_id="a1", turn_index=1, content="hello", tool_calls=[{"name": "search"}])
|
|
assert r.agent_id == "a1"
|
|
assert r.error is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_latest_ai_text_extraction(self, store):
|
|
svc = _svc(store)
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
messages = [
|
|
{"type": "human", "content": "hello"},
|
|
{"type": "ai", "content": "", "tool_calls": [{"name": "search", "args": {}, "id": "1"}]},
|
|
{"type": "ai", "content": "The answer is 42."},
|
|
]
|
|
assert adapter._latest_ai_text(messages) == "The answer is 42."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_calls_since_last_human(self, store):
|
|
svc = _svc(store)
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
messages = [
|
|
{"type": "human", "content": "old"},
|
|
{"type": "ai", "content": "", "tool_calls": [{"name": "old_tool", "args": {}, "id": "0"}]},
|
|
{"type": "human", "content": "new"},
|
|
{"type": "ai", "content": "", "tool_calls": [{"name": "search", "args": {"q": "x"}, "id": "1"}]},
|
|
]
|
|
calls = adapter._tool_calls_since_last_human(messages)
|
|
assert len(calls) == 1
|
|
assert calls[0]["name"] == "search"
|
|
|
|
|
|
# ── seat_runs persistence tests ───────────────────────────────────────────────
|
|
|
|
|
|
class TestSeatRunPersistence:
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_lifecycle(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
sr = await svc.store.seat_runs.create_seat_run(
|
|
maf_run_id=run["id"], agent_id="expert-a", turn_index=1, deerflow_thread_id="t1"
|
|
)
|
|
assert sr["status"] == "pending"
|
|
|
|
await svc.store.seat_runs.mark_running(sr["id"], deerflow_run_id="df-1")
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
assert len(active) == 1
|
|
assert active[0]["deerflow_run_id"] == "df-1"
|
|
|
|
await svc.store.seat_runs.mark_completed(sr["id"], checkpoint_id="cp-1")
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
assert len(active) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_all_active(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
sr1 = await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
sr2 = await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a2", turn_index=1)
|
|
await svc.store.seat_runs.mark_running(sr1["id"], deerflow_run_id="df-1")
|
|
await svc.store.seat_runs.mark_running(sr2["id"], deerflow_run_id="df-2")
|
|
|
|
cancelled = await svc.store.seat_runs.cancel_all_active(run["id"])
|
|
# Two-phase cancel: returns {seat_run_id, deerflow_run_id} dicts and
|
|
# leaves rows in 'cancelling' (caller finalizes to cancelled).
|
|
assert {c["deerflow_run_id"] for c in cancelled} == {"df-1", "df-2"}
|
|
assert {c["seat_run_id"] for c in cancelled} == {sr1["id"], sr2["id"]}
|
|
# 'cancelling' is not an active status, so active count drops to 0.
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
assert len(active) == 0
|
|
# Rows are mid-cancel, not yet finalized.
|
|
all_runs = await svc.store.seat_runs.get_by_maf_run(run["id"])
|
|
assert {r["status"] for r in all_runs} == {"cancelling"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unique_constraint(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
with pytest.raises(Exception):
|
|
await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
|
|
|
|
# ── event flow integration tests ──────────────────────────────────────────────
|
|
|
|
|
|
class TestAdapterEventFlow:
|
|
@pytest.mark.asyncio
|
|
async def test_full_event_sequence_persisted(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "analyze"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
await adapter._emit_critical(run["id"], make_turn_started("expert", 1))
|
|
await adapter._emit(run["id"], {"event_type": "tool.requested", "agent_id": "expert", "turn_index": 1, "payload": {"tool_name": "search"}})
|
|
await adapter._emit(run["id"], {"event_type": "tool.completed", "agent_id": "expert", "turn_index": 1, "payload": {"tool_name": "search", "status": "succeeded"}})
|
|
await adapter._emit_critical(run["id"], make_turn_completed("expert", 1, content="done"))
|
|
await adapter._emit_critical(run["id"], make_run_completed("expert", 1))
|
|
|
|
events, latest = await svc.list_events(run_id=run["id"], user_id="u1")
|
|
assert len(events) == 5
|
|
assert latest == 5
|
|
types = [e["event_type"] for e in events]
|
|
assert types == ["agent.turn_started", "tool.requested", "tool.completed", "agent.message.completed", "run.completed"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedup_via_source_event_id(self, store):
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
ev = {"event_type": "tool.completed", "agent_id": "a1", "turn_index": 1, "payload": {"tool_name": "x"}, "source_event_id": "tool:tc1"}
|
|
await adapter._emit(run["id"], ev)
|
|
await adapter._emit(run["id"], ev)
|
|
|
|
events, _ = await svc.list_events(run_id=run["id"], user_id="u1")
|
|
assert len(events) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_critical_passes_persisted_audit_id_to_bridge(self, store):
|
|
"""_emit_critical must capture the persisted audit row id and pass it
|
|
as source_audit_event_id to _bridge_conversation_event, so projection
|
|
dedup is keyed on the real audit row — not an ephemeral input field."""
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
bridge_calls: list[dict] = []
|
|
original_bridge = adapter._bridge_conversation_event
|
|
|
|
async def capture_bridge(run_id, event, *, source_audit_event_id=None):
|
|
bridge_calls.append({
|
|
"run_id": run_id,
|
|
"event_type": event.get("event_type"),
|
|
"source_audit_event_id": source_audit_event_id,
|
|
})
|
|
await original_bridge(run_id, event, source_audit_event_id=source_audit_event_id)
|
|
|
|
adapter._bridge_conversation_event = capture_bridge
|
|
|
|
critical_event = make_turn_started("expert", 1)
|
|
await adapter._emit_critical(run["id"], critical_event)
|
|
|
|
assert len(bridge_calls) == 1
|
|
assert bridge_calls[0]["source_audit_event_id"] is not None
|
|
assert isinstance(bridge_calls[0]["source_audit_event_id"], str)
|
|
assert len(bridge_calls[0]["source_audit_event_id"]) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_passes_persisted_audit_id_for_non_critical(self, store):
|
|
"""_emit (non-critical path) must also pass persisted audit row id."""
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
bridge_calls: list[dict] = []
|
|
original_bridge = adapter._bridge_conversation_event
|
|
|
|
async def capture_bridge(run_id, event, *, source_audit_event_id=None):
|
|
bridge_calls.append({
|
|
"source_audit_event_id": source_audit_event_id,
|
|
})
|
|
await original_bridge(run_id, event, source_audit_event_id=source_audit_event_id)
|
|
|
|
adapter._bridge_conversation_event = capture_bridge
|
|
|
|
ev = {
|
|
"event_type": "agent.message.delta",
|
|
"agent_id": "a1",
|
|
"turn_index": 0,
|
|
"payload": {"text": "hello"},
|
|
"source_event_id": "stream:evt_001",
|
|
}
|
|
await adapter._emit(run["id"], ev)
|
|
|
|
assert len(bridge_calls) == 1
|
|
assert bridge_calls[0]["source_audit_event_id"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_critical_dedup_across_simulated_restart(self, store):
|
|
"""Simulate: emit critical event, then 'restart' and re-emit the same
|
|
event via reconcile. The conversation projection must be deduped via
|
|
the persisted audit row id — no duplicate conversation events."""
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub
|
|
from app.maf_roundtable.conversation_service import MafConversationService
|
|
|
|
class FakeMafConfig:
|
|
enabled = True
|
|
default_mode = "structured"
|
|
default_model_name = "test"
|
|
default_reasoning_effort = None
|
|
default_manager_agent_id = "mgr"
|
|
default_max_rounds = 8
|
|
default_write_policy = "confirm_high_risk"
|
|
admin_diagnostic_access = True
|
|
|
|
hub = ConversationEventHub(heartbeat_interval=0.3)
|
|
conv_svc = MafConversationService(store, maf_config=FakeMafConfig(), event_hub=hub)
|
|
|
|
conv = await conv_svc.create_conversation(user_id="u1", title="RestartDedup")
|
|
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
svc = _svc(store)
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
|
|
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
|
|
sess_row = (
|
|
await db.execute(
|
|
select(MafRoundtableSessionRow).where(
|
|
MafRoundtableSessionRow.id == session["id"]
|
|
)
|
|
)
|
|
).scalars().first()
|
|
conv_row.internal_session_id = sess_row.id
|
|
run_row = (
|
|
await db.execute(
|
|
select(MafRoundtableRunRow).where(
|
|
MafRoundtableRunRow.id == run["id"]
|
|
)
|
|
)
|
|
).scalars().first()
|
|
assert run_row is not None
|
|
|
|
client = _mock_client()
|
|
adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1")
|
|
|
|
critical_event = {
|
|
"event_type": "action.waiting_approval",
|
|
"agent_id": "seat_a",
|
|
"turn_index": 1,
|
|
"payload": {"tool_name": "write_file", "summary": "high risk"},
|
|
}
|
|
await adapter._emit_critical(run["id"], critical_event)
|
|
|
|
await conv_svc.reconcile_conversation_events(run_id=run["id"])
|
|
|
|
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
|
|
|
|
hub.shutdown()
|