"""Phase-3 acceptance tests (P0-2): real seat-run identity & idempotency. These tests use a REAL SQLite ``MafSeatRunRepository`` (no mocks on the repository) to exercise the Phase-3 four-column unique constraint ``(maf_run_id, agent_id, turn_index, execution_stage)`` and the idempotent ``get_or_create_seat_run`` acquisition. They prove the core Phase-3 principle: The business round ``turn_index`` never changes, but every distinct DeerFlow execution gets its own persistent identity via ``execution_stage``. Coverage: 1. manager planner → cross_review → finalizer within ONE business round coexist (same agent, same turn, different stage). 2. expert → format_repair within one business round coexist (same agent, same turn, different stage). 3. The same four-tuple second insert is rejected by the DB constraint. 4. ``get_or_create_seat_run`` returns the existing row (created=False) instead of launching a duplicate when one already exists. 5. Process recovery: an already-completed stage is NOT re-created. 6. Idempotent Phase-3 migration + old rows back-filled to ``legacy``. """ from __future__ import annotations import importlib.util from datetime import datetime from pathlib import Path import pytest import pytest_asyncio from sqlalchemy import inspect as sa_inspect from sqlalchemy import text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.persistence.maf_roundtable.model import Base from deerflow.persistence.maf_roundtable.seat_runs import MafSeatRunRepository # ── fixtures ──────────────────────────────────────────────────────────────── @pytest_asyncio.fixture async def repo(): """A real SQLite-backed MafSeatRunRepository over the full schema.""" 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) # Seed the parent run row so the FK on seat_runs is satisfiable. from deerflow.persistence.maf_roundtable.model import ( MafRoundtableRunRow, MafRoundtableSessionRow, ) now = datetime(2026, 1, 1) async with sf() as session: session.add(MafRoundtableSessionRow( id="s1", user_id="u1", title="t", task_text="t", mode="structured", status="draft", write_policy="confirm_high_risk", max_rounds=8, created_at=now, updated_at=now, )) session.add(MafRoundtableRunRow( id="r1", session_id="s1", status="pending", created_at=now, updated_at=now, )) await session.commit() r = MafSeatRunRepository(sf) yield r await engine.dispose() # ── 1. Same business round, different stages coexist (manager path) ───────── @pytest.mark.asyncio async def test_manager_stages_coexist_within_one_business_round(repo): """planner / cross_review / finalizer all share (run, manager, turn) but differ by execution_stage — all three must insert without conflict.""" common = dict(maf_run_id="r1", agent_id="manager-1", turn_index=1) await repo.create_seat_run(execution_stage="planner", **common) await repo.create_seat_run(execution_stage="cross_review", **common) await repo.create_seat_run(execution_stage="finalizer", **common) rows = await repo.get_by_maf_run("r1") stages = {r["execution_stage"] for r in rows} assert stages == {"planner", "cross_review", "finalizer"} # All share the SAME business turn_index (never mutated to dodge collision). assert {r["turn_index"] for r in rows} == {1} # ── 2. expert + format_repair coexist within one business round ───────────── @pytest.mark.asyncio async def test_expert_and_format_repair_coexist(repo): """An expert seat turn and its one-shot format-repair retry share the business round but are distinct DeerFlow executions.""" common = dict(maf_run_id="r1", agent_id="expert-1", turn_index=1) await repo.create_seat_run(execution_stage="expert", **common) # Same agent, same turn, DIFFERENT stage — must NOT conflict. await repo.create_seat_run(execution_stage="format_repair", **common) rows = await repo.get_by_maf_run("r1") assert {r["execution_stage"] for r in rows} == {"expert", "format_repair"} # ── 3. Same four-tuple second insert is rejected ──────────────────────────── @pytest.mark.asyncio async def test_duplicate_four_tuple_rejected(repo): """Inserting the exact same (run, agent, turn, stage) twice must raise — the four-column unique constraint enforces at-most-one DeerFlow execution per stage per business round.""" common = dict(maf_run_id="r1", agent_id="a1", turn_index=2, execution_stage="expert") await repo.create_seat_run(**common) with pytest.raises(IntegrityError): await repo.create_seat_run(**common) # ── 4. get_or_create_seat_run is idempotent ───────────────────────────────── @pytest.mark.asyncio async def test_get_or_create_returns_existing_not_duplicate(repo): """When a seat-run for the four-tuple already exists, get_or_create must return it with created=False rather than launching a duplicate.""" key = dict(maf_run_id="r1", agent_id="a2", turn_index=3, execution_stage="cross_review") row1, created1 = await repo.get_or_create_seat_run(**key) assert created1 is True assert row1["execution_stage"] == "cross_review" # Second acquisition — existing row returned, NO duplicate created. row2, created2 = await repo.get_or_create_seat_run(**key) assert created2 is False assert row2["id"] == row1["id"] rows = await repo.get_by_maf_run("r1") cross_review_rows = [r for r in rows if r["agent_id"] == "a2"] assert len(cross_review_rows) == 1 # exactly one, no duplicate # ── 5. Process recovery: completed stage is NOT re-created ───────────────── @pytest.mark.asyncio async def test_recovery_completed_stage_not_recreated(repo): """After a simulated restart, acquiring the same four-tuple returns the already-persisted row (created=False). The caller must replay the checkpoint's saved stage result instead of starting a new DeerFlow run.""" key = dict(maf_run_id="r1", agent_id="a3", turn_index=4, execution_stage="finalizer") row, created = await repo.get_or_create_seat_run(**key) assert created is True await repo.mark_completed(row["id"], checkpoint_id="ckpt-xyz") # Simulated restart: fresh acquisition of the same stage. row2, created2 = await repo.get_or_create_seat_run(**key) assert created2 is False assert row2["id"] == row["id"] assert row2["status"] == "completed" assert row2["workflow_checkpoint_id"] == "ckpt-xyz" @pytest.mark.asyncio async def test_get_by_four_tuple_lookup(repo): """Direct four-tuple lookup returns the row or None.""" key = dict(maf_run_id="r1", agent_id="a4", turn_index=5, execution_stage="expert") assert await repo.get_by_four_tuple(**key) is None await repo.create_seat_run(**key) found = await repo.get_by_four_tuple(**key) assert found is not None assert found["agent_id"] == "a4" assert found["execution_stage"] == "expert" # ── 6. Phase-3 migration: idempotent + back-fills legacy ──────────────────── def _load_phase3_migration(): here = Path(__file__).resolve().parent.parent script = here / "scripts" / "migrate_maf_roundtable_phase3.py" spec = importlib.util.spec_from_file_location("migrate_maf_roundtable_phase3", script) assert spec and spec.loader, "could not load phase3 migration script" mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod @pytest.mark.asyncio async def test_phase3_migration_idempotent_and_backfills_legacy(): """Run the Phase-3 migration against a Phase-2 schema (old 3-tuple constraint), then re-run it (idempotent). Old seat-run rows are back-filled with execution_stage='legacy'.""" mod = _load_phase3_migration() engine = create_async_engine("sqlite+aiosqlite://", echo=False) # Build the OLD Phase-2 seat_runs table (3-tuple unique, NO execution_stage). async with engine.begin() as conn: await conn.execute(text(""" CREATE TABLE maf_roundtable_runs ( id VARCHAR(64) NOT NULL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, status VARCHAR(32) NOT NULL, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ) """)) await conn.execute(text(""" CREATE TABLE maf_roundtable_sessions ( id VARCHAR(64) NOT NULL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, title VARCHAR(255), task_text TEXT, mode VARCHAR(32), status VARCHAR(32), write_policy VARCHAR(32), max_rounds INTEGER, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ) """)) await conn.execute(text(""" CREATE TABLE maf_roundtable_seat_runs ( id VARCHAR(64) NOT NULL PRIMARY KEY, maf_run_id VARCHAR(64) NOT NULL, agent_id VARCHAR(64) NOT NULL, turn_index INTEGER NOT NULL, deerflow_run_id VARCHAR(64), deerflow_thread_id VARCHAR(64), sandbox_id VARCHAR(64), workflow_checkpoint_id VARCHAR(128), status VARCHAR(32) DEFAULT 'pending' NOT NULL, started_at DATETIME, finished_at DATETIME, error TEXT, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, FOREIGN KEY (maf_run_id) REFERENCES maf_roundtable_runs (id) ON DELETE CASCADE ) """)) await conn.execute(text( "CREATE UNIQUE INDEX uq_maf_seat_runs_run_agent_turn " "ON maf_roundtable_seat_runs (maf_run_id, agent_id, turn_index)" )) # Seed parent run + a pre-Phase-3 seat-run row. await conn.execute(text( "INSERT INTO maf_roundtable_sessions (id, user_id, title, task_text, mode, status, write_policy, max_rounds, created_at, updated_at) " "VALUES ('s1','u1','t','t','structured','draft','confirm_high_risk',8,'2026-01-01','2026-01-01')" )) await conn.execute(text( "INSERT INTO maf_roundtable_runs (id, session_id, status, created_at, updated_at) " "VALUES ('r1','s1','pending','2026-01-01','2026-01-01')" )) await conn.execute(text( "INSERT INTO maf_roundtable_seat_runs (id, maf_run_id, agent_id, turn_index, status, created_at, updated_at) " "VALUES ('sr1','r1','a1',1,'completed','2026-01-01','2026-01-01')" )) # Run the migration. async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) sf = async_sessionmaker(engine, expire_on_commit=False) def _verify(sync_conn): insp = sa_inspect(sync_conn) cols = {c["name"] for c in insp.get_columns("maf_roundtable_seat_runs")} assert "execution_stage" in cols # The NEW four-column unique constraint/index is present. new_constraint = mod._find_unique_constraint( insp, "maf_roundtable_seat_runs", expected_cols={"maf_run_id", "agent_id", "turn_index", "execution_stage"}, ) assert new_constraint is not None, "new four-column unique constraint missing" # The OLD three-column constraint must be gone (rebuilt in SQLite). old_constraint = mod._find_unique_constraint( insp, "maf_roundtable_seat_runs", expected_cols={"maf_run_id", "agent_id", "turn_index"}, ) assert old_constraint is None, "old three-column constraint still present after migration" async with engine.begin() as conn: await conn.run_sync(_verify) # Old row back-filled to 'legacy'. async with sf() as session: row = (await session.execute( text("SELECT execution_stage FROM maf_roundtable_seat_runs WHERE id='sr1'") )).one_or_none() assert row is not None assert row[0] == "legacy" # Re-run the migration — must be idempotent (no error). async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=True, logger=mod.logger, changes_ref=[0]) # After migration, the store can create rows using the new column, and the # constraint allows cross_review + finalizer to coexist for the same # business round (the regression the migration fixes). store = make_maf_roundtable_store(sf) await store.seat_runs.create_seat_run( maf_run_id="r1", agent_id="mgr", turn_index=1, execution_stage="cross_review" ) await store.seat_runs.create_seat_run( maf_run_id="r1", agent_id="mgr", turn_index=1, execution_stage="finalizer" ) rows = await store.seat_runs.get_by_maf_run("r1") new_mgr_stages = {r["execution_stage"] for r in rows if r["agent_id"] == "mgr"} assert new_mgr_stages == {"cross_review", "finalizer"} await engine.dispose()