"""Tests for the MAF roundtable persistence + service layer (?????). Covers the Phase 1 exit criteria: - Session CRUD with participants + capability snapshot. - Run lifecycle (create / get / list / cancel). - Monotonic per-run event seq under concurrent appends (MAF-01-06). - Event ???? (list_events after_seq=N). - Action idempotency-key uniqueness. - Approval decision transition + "already decided" guard. - User isolation (MAF-01-07) ??non-owner gets None / 404, not 403. - Admin diagnostic access bypass. - Feature flag (MAF-01-04) ??writes raise MafDisabledError when off; reads open. - Run gating ??running session rejects config edits. - capability_snapshot frozen at creation. - MAF tables have no FK to legacy roundtable_* tables. """ from __future__ import annotations import asyncio 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.service import ( MafConflictError, MafNotFoundError, ) from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store @pytest_asyncio.fixture async def store(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf.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() def _svc(store, *, enabled=True, admin_diagnostic_access=True): return make_maf_service(store, enabled=enabled, admin_diagnostic_access=admin_diagnostic_access) @pytest.mark.asyncio async def test_create_and_get_session_round_trips(store): svc = _svc(store) created = await svc.create_session( user_id="u1", body={ "title": "????", "task_text": "????????", "mode": "structured", "model_name": "glm-5", "max_rounds": 5, "write_policy": "confirm_high_risk", "participants": [ {"agent_id": "mgr", "display_name": "host", "role": "manager", "sort_order": 0}, {"agent_id": "seat-a", "display_name": "?A", "role": "seat", "sort_order": 1}, ], }, ) assert created["user_id"] == "u1" assert created["status"] == "draft" assert created["mode"] == "structured" snap = created["capability_snapshot"] assert snap is not None assert snap.get("version") == 1 assert isinstance(snap.get("agents"), list) assert len(snap["agents"]) == 2 assert len(created["participants"]) == 2 assert created["participants"][0]["agent_id"] == "mgr" fetched = await svc.get_session(session_id=created["id"], user_id="u1") assert fetched["id"] == created["id"] assert len(fetched["participants"]) == 2 @pytest.mark.asyncio async def test_list_sessions_owner_scoped(store): svc = _svc(store) await svc.create_session(user_id="u1", body={"task_text": "t1"}) await svc.create_session(user_id="u2", body={"task_text": "t2"}) rows_u1, total_u1 = await svc.list_sessions(user_id="u1") assert total_u1 == 1 assert rows_u1[0]["user_id"] == "u1" rows_u2, total_u2 = await svc.list_sessions(user_id="u2") assert total_u2 == 1 assert rows_u2[0]["user_id"] == "u2" @pytest.mark.asyncio async def test_update_session_partial_and_participants_replace(store): svc = _svc(store) s = await svc.create_session( user_id="u1", body={"task_text": "orig", "participants": [{"agent_id": "a1", "role": "seat"}]}, ) updated = await svc.update_session( session_id=s["id"], user_id="u1", fields={"title": "new-title", "participants": [{"agent_id": "a2", "role": "seat"}]}, ) assert updated["title"] == "new-title" assert updated["task_text"] == "orig" assert {p["agent_id"] for p in updated["participants"]} == {"a2"} @pytest.mark.asyncio async def test_update_session_freezes_capability_snapshot(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) original = s["capability_snapshot"] updated = await svc.update_session( session_id=s["id"], user_id="u1", fields={"capability_snapshot": {"v": 999}} ) assert updated["capability_snapshot"] == original @pytest.mark.asyncio async def test_delete_session_owner_scoped(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) with pytest.raises(MafNotFoundError): await svc.delete_session(session_id=s["id"], user_id="u2") await svc.delete_session(session_id=s["id"], user_id="u1") with pytest.raises(MafNotFoundError): await svc.get_session(session_id=s["id"], user_id="u1") # ?? run lifecycle ?????????????????????????????????????????????????????????? @pytest.mark.asyncio async def test_create_run_sets_session_running(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") # No real MAF executor exists yet (Phase 2+): the run is parked in # ``pending`` (a clearly reserved, not-yet-executing state) rather than # pretending to be ``running``; the session is occupied (``running``). assert run["status"] == "pending" refreshed = await svc.get_session(session_id=s["id"], user_id="u1") assert refreshed["status"] == "running" @pytest.mark.asyncio async def test_create_run_rejects_second_active_run(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) await svc.create_run(session_id=s["id"], user_id="u1") with pytest.raises(MafConflictError): await svc.create_run(session_id=s["id"], user_id="u1") @pytest.mark.asyncio async def test_update_session_rejected_while_running(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) await svc.create_run(session_id=s["id"], user_id="u1") with pytest.raises(MafConflictError): await svc.update_session(session_id=s["id"], user_id="u1", fields={"title": "x"}) @pytest.mark.asyncio async def test_cancel_run_marks_session_cancelled(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1") assert cancelled["status"] == "cancelled" refreshed = await svc.get_session(session_id=s["id"], user_id="u1") assert refreshed["status"] == "cancelled" # ?? events: monotonic seq + concurrent appends (MAF-01-06) ???????????????? @pytest.mark.asyncio async def test_append_event_monotonic_seq(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") e1 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="run.started") e2 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="agent.message.completed") e3 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="run.completed") assert [e1["seq"], e2["seq"], e3["seq"]] == [1, 2, 3] refreshed_run = await svc.get_run(run_id=run["id"], user_id="u1") assert refreshed_run["last_event_seq"] == 3 @pytest.mark.asyncio async def test_concurrent_appends_produce_dense_seq_set(store): """Many parallel appends must produce a dense 1..N seq set ??no gaps, no collisions ??exercising the MAX+1 retry on the (run_id, seq) unique constraint (MAF-01-06).""" svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") n = 40 results = await asyncio.gather( *[ svc.append_event( run_id=run["id"], user_id="u1", event_type="agent.message.delta", payload={"i": i}, ) for i in range(n) ] ) seqs = sorted(r["seq"] for r in results) assert seqs == list(range(1, n + 1)) @pytest.mark.asyncio async def test_list_events_after_seq_pull(store): """????: events with seq > after_seq in ascending order.""" svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") for i in range(5): await svc.append_event(run_id=run["id"], user_id="u1", event_type="tool.progress", payload={"i": i}) events, latest = await svc.list_events(run_id=run["id"], user_id="u1", after_seq=2) assert [e["seq"] for e in events] == [3, 4, 5] assert latest == 5 # ?? actions: idempotency key ?????????????????????????????????????????????? @pytest.mark.asyncio async def test_action_idempotency_key_unique(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") from sqlalchemy.exc import IntegrityError await store.actions.create_action( { "session_id": s["id"], "run_id": run["id"], "tool_name": "update_task", "idempotency_key": "sess1+run1+seat-a+update_task+act1", "risk_level": "R2", "status": "pending", } ) with pytest.raises(IntegrityError): await store.actions.create_action( { "session_id": s["id"], "run_id": run["id"], "tool_name": "update_task", "idempotency_key": "sess1+run1+seat-a+update_task+act1", "status": "pending", } ) found = await store.actions.get_by_idempotency_key("sess1+run1+seat-a+update_task+act1") assert found is not None assert found["tool_name"] == "update_task" # ?? approvals: decision transition ???????????????????????????????????????? @pytest.mark.asyncio async def test_approval_decision_and_already_decided_guard(store): svc = _svc(store) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") appr = await store.approvals.create_approval( {"session_id": s["id"], "run_id": run["id"], "summary": "approve write", "status": "pending"} ) decided = await svc.decide_approval(approval_id=appr["id"], decision="approved", user_id="u1") assert decided["status"] == "approved" assert decided["decided_by"] == "u1" with pytest.raises(MafConflictError): await svc.decide_approval(approval_id=appr["id"], decision="rejected", user_id="u1")