deerflow-code/offline-backend-20260512/backend/tests/test_maf_phase3_seat_runs.py

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