620 lines
25 KiB
Python
620 lines
25 KiB
Python
"""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()
|