"""Phase-3 Round-2 acceptance tests (P0-1): race-free idempotency & replay.

These prove the Phase-3 promise: a distinct DeerFlow execution (run, agent,
turn, execution_stage) is started at most once, even under concurrency,
retries, recovery, or a partial crash. They use a REAL SQLite-backed store +
the REAL ``DeerFlowAgentAdapter`` (only ``_execute_run`` is faked, so the
adapter's get_or_create / status-branching / snapshot logic is exercised
end-to-end against a real DB).

Coverage:
- Two coroutines acquiring the same four-tuple concurrently → exactly one wins
  (created=True), the other reads the existing row (created=False). This is the
  race that "query-then-insert" lost; the atomic insert-then-read wins it.
- A double-call of ``run_seat_turn`` after completion does NOT start a second
  DeerFlow run — it replays the saved ``result_snapshot``.
- A call while the stage is pending/running returns an in-flight marker and
  does NOT start a second DeerFlow run.
- A failed/cancelled stage is terminal and never auto-retried.
"""

from __future__ import annotations

import asyncio
import importlib.util
import logging
from datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

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.deerflow_agent_adapter import DeerFlowAgentAdapter, SeatResponse
from app.maf_roundtable.runtime_client import DeerFlowRuntimeClient
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import Base, MafRoundtableRunRow, MafRoundtableSessionRow

logger = logging.getLogger(__name__)


@pytest_asyncio.fixture
async def svc(tmp_path):
    # IMPORTANT: a file-based SQLite is required for the concurrency test. The
    # in-memory engine (``sqlite+aiosqlite://``) multiplexes all sessions over a
    # SINGLE connection, so two coroutines in ``asyncio.gather`` share one
    # transaction — the loser's rollback would void the winner's uncommitted
    # insert, making a true race impossible to simulate. A file DB gives each
    # ``async with self._sf()`` its own connection (production behavior), so the
    # four-column unique constraint is the real arbiter.
    db_path = tmp_path / "idempotency.db"
    engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    store = make_maf_roundtable_store(sf)
    service = make_maf_service(store)

    # Seed a session + run so the seat_runs FK is satisfiable.
    now = datetime(2026, 1, 1)
    async with sf() as session:
        session.add(MafRoundtableSessionRow(
            id="sess-1", user_id="u1", title="t", task_text="t",
            mode="structured", status="running", write_policy="auto",
            max_rounds=8, created_at=now, updated_at=now,
        ))
        session.add(MafRoundtableRunRow(
            id="run-1", session_id="sess-1", status="running",
            created_at=now, updated_at=now,
        ))
        await session.commit()

    yield service
    await engine.dispose()


def _adapter(svc) -> DeerFlowAgentAdapter:
    app = MagicMock()
    app.state.checkpointer = AsyncMock()
    app.state.thread_store = AsyncMock()
    app.state.stream_bridge = None
    app.state.run_manager = AsyncMock()
    client = DeerFlowRuntimeClient(app)
    return DeerFlowAgentAdapter(client, service=svc, user_id="u1")


# ── 1. Concurrent acquisition: exactly one wins ─────────────────────────────


@pytest.mark.asyncio
async def test_concurrent_get_or_create_one_winner(svc):
    """Two coroutines acquire the same four-tuple at (near-)the same time. The
    atomic insert-then-read means the DB constraint picks a single winner; the
    loser reads the winner's row. Both calls return, neither raises, and there
    is exactly ONE row for the four-tuple."""
    repo = svc.store.seat_runs
    key = dict(maf_run_id="run-1", agent_id="a1", turn_index=1, execution_stage="expert")

    (ar, ac), (br, bc) = await asyncio.gather(
        repo.get_or_create_seat_run(**key),
        repo.get_or_create_seat_run(**key),
    )
    assert (ac, bc) in {(True, False), (False, True)}, "exactly one winner; the other reads existing"
    assert ar["id"] == br["id"]
    rows = await repo.get_by_maf_run("run-1")
    assert len([r for r in rows if r["agent_id"] == "a1" and r["execution_stage"] == "expert"]) == 1


# ── 2. Double-call after completion replays the snapshot ────────────────────


@pytest.mark.asyncio
async def test_double_call_after_completion_does_not_re_run(svc):
    """A second ``run_seat_turn`` for an already-completed stage must replay the
    saved ``result_snapshot`` and NOT call ``_execute_run`` again. This is the
    core recovery guarantee: a restart or retry never re-invokes the model."""
    adapter = _adapter(svc)
    calls = {"execute": 0}

    async def _fake_execute(**kwargs):
        calls["execute"] += 1
        return SeatResponse(
            agent_id=kwargs["agent_id"],
            turn_index=kwargs["turn_index"],
            content="first-run-output",
            tool_calls=[{"name": "search", "args": {}, "id": "tc1"}],
            usage={"input_tokens": 5, "output_tokens": 5},
            metadata={"thread_id": kwargs["thread_id"]},
        )

    adapter._execute_run = _fake_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    common = dict(
        session_id="sess-1", run_id="run-1", agent_id="a2", turn_index=2,
        execution_stage="expert", message="q", thread_id="t-a2",
        task_text="q", stage_instruction="",
    )
    first = await adapter.run_seat_turn(**common)
    assert calls["execute"] == 1
    assert first.content == "first-run-output"

    # Second call — completed stage. Must replay, NOT re-run.
    second = await adapter.run_seat_turn(**common)
    assert calls["execute"] == 1, "completed stage was re-run instead of replayed"
    assert second.content == "first-run-output"
    assert second.error is None


# ── 3. In-flight stage: second caller waits and replays ─────────────────────


@pytest.mark.asyncio
async def test_running_stage_second_caller_waits_and_replays(svc):
    """When a stage is pending/running (another caller owns it), a second
    ``run_seat_turn`` must WAIT for the first to complete, then replay the
    same snapshot. ``_execute_run`` is called exactly once."""
    adapter = _adapter(svc)
    execute_started = asyncio.Event()
    release_execute = asyncio.Event()
    calls = {"execute": 0}

    async def _blocking_execute(**kwargs):
        calls["execute"] += 1
        execute_started.set()
        await release_execute.wait()
        return SeatResponse(
            agent_id=kwargs["agent_id"], turn_index=kwargs["turn_index"],
            content="done-content", tool_calls=[{"name": "t", "args": {}, "id": "x"}],
            usage={"input_tokens": 1}, metadata={"thread_id": kwargs["thread_id"]},
        )

    adapter._execute_run = _blocking_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    common = dict(
        session_id="sess-1", run_id="run-1", agent_id="a3", turn_index=3,
        execution_stage="cross_review", message="q", thread_id="t-a3",
        task_text="q", stage_instruction="",
    )
    first_task = asyncio.create_task(adapter.run_seat_turn(**common))
    await execute_started.wait()

    second_task = asyncio.create_task(adapter.run_seat_turn(**common, wait_timeout_seconds=10))
    await asyncio.sleep(0.05)

    release_execute.set()
    first = await first_task
    second = await second_task

    assert calls["execute"] == 1, "in-flight stage started a second DeerFlow run"
    assert first.content == "done-content"
    assert second.content == "done-content"
    assert second.error is None


# ── 4. Failed/cancelled stage is terminal ────────────────────────────────────


@pytest.mark.asyncio
async def test_failed_stage_not_auto_retried(svc):
    """A previously-failed stage is terminal; a new ``run_seat_turn`` surfaces
    a terminal marker and never auto-retries (no silent re-run)."""
    adapter = _adapter(svc)
    calls = {"execute": 0}

    async def _failing_execute(**kwargs):
        calls["execute"] += 1
        raise RuntimeError("model blew up")

    adapter._execute_run = _failing_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    common = dict(
        session_id="sess-1", run_id="run-1", agent_id="a4", turn_index=4,
        execution_stage="finalizer", message="q", thread_id="t-a4",
        task_text="q", stage_instruction="",
    )
    first = await adapter.run_seat_turn(**common)
    assert first.error  # failed
    assert calls["execute"] == 1

    second = await adapter.run_seat_turn(**common)
    assert calls["execute"] == 1, "failed stage was auto-retried"
    assert second.error is not None
    assert second.metadata is not None and second.metadata.get("terminal") == "failed"


# ── 5. Four-tuple replay snapshot persists sanitized content ─────────────────


@pytest.mark.asyncio
async def test_result_snapshot_persisted_on_completion(svc):
    """The sanitized result snapshot is written atomically with the completed
    transition so recovery can replay it even if the workflow checkpoint was
    not yet written."""
    adapter = _adapter(svc)

    async def _fake_execute(**kwargs):
        return SeatResponse(
            agent_id=kwargs["agent_id"], turn_index=kwargs["turn_index"],
            content="snapshot-me", tool_calls=[{"name": "t", "args": {}, "id": "x"}],
            usage={"input_tokens": 1}, metadata={"thread_id": "t"},
        )

    adapter._execute_run = _fake_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    await adapter.run_seat_turn(
        session_id="sess-1", run_id="run-1", agent_id="a5", turn_index=5,
        execution_stage="expert", message="q", thread_id="t-a5",
        task_text="q", stage_instruction="",
    )

    row = await svc.store.seat_runs.get_by_four_tuple(
        maf_run_id="run-1", agent_id="a5", turn_index=5, execution_stage="expert",
    )
    assert row is not None
    assert row["status"] == "completed"
    snap = row["result_snapshot"]
    assert snap is not None
    assert snap["content"] == "snapshot-me"
    assert snap["tool_calls"][0]["name"] == "t"


# ── 6. Invalid execution_stage rejected ──────────────────────────────────────


@pytest.mark.asyncio
async def test_invalid_execution_stage_rejected(svc):
    """A typo'd execution_stage (e.g. 'exprt') is rejected at the repository —
    it would otherwise bypass the four-column idempotency key by creating a
    brand-new stage."""
    from deerflow.persistence.maf_roundtable.seat_runs import InvalidExecutionStageError

    repo = svc.store.seat_runs
    with pytest.raises(InvalidExecutionStageError):
        await repo.get_or_create_seat_run(
            maf_run_id="run-1", agent_id="a6", turn_index=6,
            execution_stage="exprt",  # typo
        )
    with pytest.raises(InvalidExecutionStageError):
        await repo.create_seat_run(
            maf_run_id="run-1", agent_id="a6", turn_index=6,
            execution_stage="expert2",  # not in the closed enum
        )


# ── 7. Migration branching: constraint vs index drop ─────────────────────────


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


def test_migration_ddl_drop_branches_on_kind():
    """The migration must emit DROP CONSTRAINT for a table constraint and DROP
    INDEX for a unique index — the Phase-2 deployment created the constraint as
    a CREATE UNIQUE INDEX, so calling DROP CONSTRAINT on it failed."""
    mod = _load_phase3_migration()
    # A constraint-kind object → DROP CONSTRAINT.
    drop_constraint_sql = mod._ddl_drop_constraint("postgresql", "uq_maf_seat_runs_run_agent_turn")
    assert "DROP CONSTRAINT" in drop_constraint_sql
    # An index-kind object → DROP INDEX.
    drop_index_sql = mod._ddl_drop_index("uq_maf_seat_runs_run_agent_turn")
    assert "DROP INDEX" in drop_index_sql


@pytest.mark.asyncio
async def test_migration_sqlite_rebuild_restores_deerflow_run_id_index():
    """The SQLite rebuild must recreate the ``deerflow_run_id`` index — without
    it, cancel-by-deerflow-run-id lookups silently degrade after a restart."""
    mod = _load_phase3_migration()
    rebuild = mod._sqlite_rebuild_seat_runs(source_has_stage=False, source_has_snapshot=False)
    joined = "\n".join(rebuild)
    assert "deerflow_run_id" in joined, "rebuild dropped the deerflow_run_id index"
    assert "ck_maf_seat_runs_execution_stage" in joined
    assert "execution_stage IN" in joined


# ── 8. P0-1: Two adapters same stage — wait, not partial_failure ─────────────


@pytest.mark.asyncio
async def test_two_adapters_same_stage_wait_not_partial_failure(svc):
    """Two adapter calls for the same four-tuple: the first blocks executing,
    the second waits. After release, both get the same snapshot. _execute_run
    is called exactly once. Neither sets partial_failure."""
    adapter = _adapter(svc)
    execute_started = asyncio.Event()
    release_execute = asyncio.Event()
    calls = {"execute": 0}

    async def _blocking_execute(**kwargs):
        calls["execute"] += 1
        execute_started.set()
        await release_execute.wait()
        return SeatResponse(
            agent_id=kwargs["agent_id"], turn_index=kwargs["turn_index"],
            content="shared-result", tool_calls=[], usage={"input_tokens": 2},
            metadata={"thread_id": kwargs["thread_id"]},
        )

    adapter._execute_run = _blocking_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    common = dict(
        session_id="sess-1", run_id="run-1", agent_id="a-wait", turn_index=7,
        execution_stage="expert", message="q", thread_id="t-wait",
        task_text="q", stage_instruction="",
    )
    task1 = asyncio.create_task(adapter.run_seat_turn(**common))
    await execute_started.wait()

    task2 = asyncio.create_task(adapter.run_seat_turn(**common, wait_timeout_seconds=10))
    await asyncio.sleep(0.05)

    release_execute.set()
    r1, r2 = await asyncio.gather(task1, task2)

    assert calls["execute"] == 1
    assert r1.content == "shared-result"
    assert r2.content == "shared-result"
    assert r1.error is None
    assert r2.error is None


# ── 9. P0-1: Empty snapshot defense ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_completed_with_empty_snapshot_returns_error(svc):
    """A completed seat run with no result_snapshot must NOT produce a fake
    success — it returns an explicit error requiring manual recovery."""
    adapter = _adapter(svc)
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    seat_run = await svc.store.seat_runs.create_seat_run(
        maf_run_id="run-1", agent_id="a-empty", turn_index=8, execution_stage="expert",
    )
    await svc.store.seat_runs.mark_completed(seat_run["id"], result_snapshot=None)

    result = await adapter.run_seat_turn(
        session_id="sess-1", run_id="run-1", agent_id="a-empty", turn_index=8,
        execution_stage="expert", message="q", thread_id="t-empty",
        task_text="q", stage_instruction="",
    )
    assert result.error is not None
    assert "no replayable snapshot" in result.error
    assert result.content == ""


# ── 10. P0-2: Snapshot sanitization before persist ───────────────────────────


@pytest.mark.asyncio
async def test_result_snapshot_sanitized_before_persist(svc):
    """Secrets in tool_calls args and metadata must be redacted in the
    persisted snapshot. Safe fields are preserved."""
    adapter = _adapter(svc)

    async def _fake_execute(**kwargs):
        return SeatResponse(
            agent_id=kwargs["agent_id"], turn_index=kwargs["turn_index"],
            content="analysis complete",
            tool_calls=[{
                "name": "request_api",
                "args": {
                    "api_key": "secret-api-key",
                    "password": "secret-password",
                    "normal": "safe",
                },
                "id": "tc1",
            }],
            usage={"input_tokens": 1},
            metadata={"authorization": "Bearer secret-token", "thread_id": "t"},
        )

    adapter._execute_run = _fake_execute  # type: ignore[method-assign]
    adapter._emit_critical = AsyncMock()  # type: ignore[method-assign]
    adapter._emit = AsyncMock()  # type: ignore[method-assign]

    await adapter.run_seat_turn(
        session_id="sess-1", run_id="run-1", agent_id="a-san", turn_index=9,
        execution_stage="expert", message="q", thread_id="t-san",
        task_text="q", stage_instruction="",
    )

    row = await svc.store.seat_runs.get_by_four_tuple(
        maf_run_id="run-1", agent_id="a-san", turn_index=9, execution_stage="expert",
    )
    assert row is not None
    snap = row["result_snapshot"]
    assert snap is not None
    snap_str = str(snap)
    assert "secret-api-key" not in snap_str
    assert "secret-password" not in snap_str
    assert "secret-token" not in snap_str
    assert snap["tool_calls"][0]["args"]["normal"] == "safe"

    replayed = await adapter.run_seat_turn(
        session_id="sess-1", run_id="run-1", agent_id="a-san", turn_index=9,
        execution_stage="expert", message="q", thread_id="t-san",
        task_text="q", stage_instruction="",
    )
    assert replayed.content == "analysis complete"
    assert "secret-api-key" not in str(replayed.tool_calls)


# ── 11. P1-1: Migration intermediate state tests ─────────────────────────────


def test_migration_rebuild_preserves_existing_stage_and_snapshot():
    """When the source table already has execution_stage and result_snapshot,
    the rebuild must copy them verbatim — not overwrite with 'legacy'/NULL."""
    mod = _load_phase3_migration()
    rebuild = mod._sqlite_rebuild_seat_runs(source_has_stage=True, source_has_snapshot=True)
    insert_sql = rebuild[1]
    assert "execution_stage" in insert_sql
    assert "'legacy'" not in insert_sql.split("SELECT")[1]
    assert "result_snapshot" in insert_sql
    assert "NULL" not in insert_sql.split("SELECT")[1].split("FROM")[0].replace("result_snapshot", "")


def test_migration_rebuild_from_phase2_uses_defaults():
    """When the source table has neither stage nor snapshot (Phase-2), the
    rebuild uses 'legacy' and NULL."""
    mod = _load_phase3_migration()
    rebuild = mod._sqlite_rebuild_seat_runs(source_has_stage=False, source_has_snapshot=False)
    insert_sql = rebuild[1]
    select_part = insert_sql.split("SELECT")[1].split("FROM")[0]
    assert "'legacy'" in select_part
    assert "NULL" in select_part


@pytest.mark.asyncio
async def test_migration_idempotent_double_run(tmp_path):
    """Running the migration twice on the same SQLite DB succeeds both times."""
    import sqlite3

    db_path = tmp_path / "double_migrate.db"
    conn = sqlite3.connect(str(db_path))
    conn.execute("""
        CREATE TABLE maf_roundtable_runs (
            id VARCHAR(64) NOT NULL PRIMARY KEY,
            session_id VARCHAR(64) NOT NULL,
            status VARCHAR(32) DEFAULT 'pending' NOT NULL,
            created_at DATETIME NOT NULL,
            updated_at DATETIME NOT NULL
        )
    """)
    conn.execute("""
        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,
            CONSTRAINT uq_maf_seat_runs_run_agent_turn UNIQUE (maf_run_id, agent_id, turn_index)
        )
    """)
    conn.execute("""
        INSERT INTO maf_roundtable_seat_runs
        (id, maf_run_id, agent_id, turn_index, status, created_at, updated_at)
        VALUES ('sr1', 'run-1', 'a1', 1, 'completed', '2026-01-01', '2026-01-01')
    """)
    conn.commit()
    conn.close()

    from sqlalchemy import create_engine, inspect, text

    engine = create_engine(f"sqlite:///{db_path}")
    mod = _load_phase3_migration()

    for _ in range(2):
        with engine.connect() as sync_conn:
            mod._do_migrate_sync(
                sync_conn, dry_run=False, logger=logger, changes_ref=[0]
            )
            sync_conn.commit()

    insp = inspect(engine)
    cols = {c["name"] for c in insp.get_columns("maf_roundtable_seat_runs")}
    assert "execution_stage" in cols
    assert "result_snapshot" in cols

    with engine.connect() as conn:
        row = conn.execute(text("SELECT execution_stage, result_snapshot FROM maf_roundtable_seat_runs WHERE id='sr1'")).fetchone()
        assert row[0] == "legacy"
    engine.dispose()


@pytest.mark.asyncio
async def test_migration_intermediate_preserves_expert_stage(tmp_path):
    """An intermediate table (has stage + 4-col unique, but no snapshot/check)
    must preserve existing execution_stage='expert' and result_snapshot data
    after rebuild."""
    import sqlite3

    db_path = tmp_path / "intermediate.db"
    conn = sqlite3.connect(str(db_path))
    conn.execute("""
        CREATE TABLE maf_roundtable_runs (
            id VARCHAR(64) NOT NULL PRIMARY KEY,
            session_id VARCHAR(64) NOT NULL,
            status VARCHAR(32) DEFAULT 'pending' NOT NULL,
            created_at DATETIME NOT NULL,
            updated_at DATETIME NOT NULL
        )
    """)
    conn.execute("""
        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,
            execution_stage VARCHAR(32) NOT NULL DEFAULT 'legacy',
            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,
            result_snapshot TEXT,
            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,
            CONSTRAINT uq_maf_seat_runs_run_agent_turn_stage UNIQUE (maf_run_id, agent_id, turn_index, execution_stage)
        )
    """)
    conn.execute("""
        INSERT INTO maf_roundtable_seat_runs
        (id, maf_run_id, agent_id, turn_index, execution_stage, status, result_snapshot, created_at, updated_at)
        VALUES ('sr2', 'run-1', 'a1', 1, 'expert', 'completed', '{"content": "keep-me"}', '2026-01-01', '2026-01-01')
    """)
    conn.commit()
    conn.close()

    from sqlalchemy import create_engine, text

    engine = create_engine(f"sqlite:///{db_path}")
    mod = _load_phase3_migration()

    with engine.connect() as sync_conn:
        mod._do_migrate_sync(sync_conn, dry_run=False, logger=logger, changes_ref=[0])
        sync_conn.commit()

    with engine.connect() as conn:
        row = conn.execute(text(
            "SELECT execution_stage, result_snapshot FROM maf_roundtable_seat_runs WHERE id='sr2'"
        )).fetchone()
        assert row[0] == "expert", f"stage corrupted to {row[0]!r}"
        assert "keep-me" in (row[1] or ""), f"snapshot corrupted to {row[1]!r}"
    engine.dispose()