551 lines
27 KiB
Python
551 lines
27 KiB
Python
"""Tests for the MAF roundtable conversation V2 migration script.
|
|
|
|
Covers:
|
|
- SQLite real migration idempotency (run twice, no error).
|
|
- Schema validation passes after migration.
|
|
- Deprecated last_message_seq column is not present.
|
|
- PostgreSQL / MySQL DDL assertions (dialect-specific SQL generation).
|
|
- Existing tables are not destroyed by re-running.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from scripts.migrate_maf_roundtable_conversation_v2 import (
|
|
CONVERSATIONS_TABLE,
|
|
EVENTS_TABLE,
|
|
MESSAGES_TABLE,
|
|
_conversations_ddl,
|
|
_events_ddl,
|
|
_events_table_has_source_unique,
|
|
_indexes_ddl,
|
|
_messages_ddl,
|
|
run_migration,
|
|
validate_schema,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sqlite_migration_creates_tables(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'mig.db'}"
|
|
await run_migration(db_url)
|
|
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
def _check(sync_conn):
|
|
insp = inspect(sync_conn)
|
|
tables = insp.get_table_names()
|
|
issues = validate_schema(insp)
|
|
return tables, issues
|
|
|
|
tables, issues = await conn.run_sync(_check)
|
|
assert CONVERSATIONS_TABLE in tables
|
|
assert MESSAGES_TABLE in tables
|
|
assert EVENTS_TABLE in tables
|
|
assert issues == [], f"Schema issues: {issues}"
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sqlite_migration_idempotent(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'mig2.db'}"
|
|
await run_migration(db_url)
|
|
await run_migration(db_url)
|
|
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
def _check(sync_conn):
|
|
return validate_schema(inspect(sync_conn))
|
|
|
|
issues = await conn.run_sync(_check)
|
|
assert issues == [], f"Schema issues after 2nd run: {issues}"
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_deprecated_last_message_seq(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'mig3.db'}"
|
|
await run_migration(db_url)
|
|
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
def _check(sync_conn):
|
|
insp = inspect(sync_conn)
|
|
return {c["name"] for c in insp.get_columns(CONVERSATIONS_TABLE)}
|
|
|
|
cols = await conn.run_sync(_check)
|
|
assert "last_stream_seq" in cols
|
|
assert "last_message_seq" not in cols
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_data_preserved(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'mig4.db'}"
|
|
await run_migration(db_url)
|
|
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
await conn.execute(text(
|
|
f"INSERT INTO {CONVERSATIONS_TABLE} (id, owner_user_id, title, safe_status, last_stream_seq, created_at, updated_at) "
|
|
"VALUES ('c1', 'u1', 'keep', 'idle', 0, '2026-01-01', '2026-01-01')"
|
|
))
|
|
|
|
await run_migration(db_url)
|
|
|
|
async with engine.begin() as conn:
|
|
row = (await conn.execute(text(f"SELECT id, title FROM {CONVERSATIONS_TABLE} WHERE id='c1'"))).first()
|
|
assert row is not None
|
|
assert row[1] == "keep"
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_partial_conversation_table_is_rebuilt_with_foreign_keys(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'partial.db'}"
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
# Represents the first uncommitted V2 draft: the table exists but its
|
|
# private run/session links were not constrained.
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_context_json TEXT, "
|
|
"safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', active_run_id VARCHAR(64), "
|
|
"internal_session_id VARCHAR(64), last_stream_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await engine.dispose()
|
|
|
|
await run_migration(db_url)
|
|
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
def _check(sync_conn):
|
|
return validate_schema(inspect(sync_conn))
|
|
|
|
assert await conn.run_sync(_check) == []
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonempty_partial_conversation_table_fails_closed(tmp_path):
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'partial-data.db'}"
|
|
engine = create_async_engine(db_url)
|
|
async with engine.begin() as conn:
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', "
|
|
"last_stream_seq INTEGER NOT NULL DEFAULT 0, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {CONVERSATIONS_TABLE} (id, owner_user_id, title, safe_status, last_stream_seq, created_at, updated_at) "
|
|
"VALUES ('c1', 'u1', 'keep', 'idle', 0, '2026-01-01', '2026-01-01')"
|
|
))
|
|
await engine.dispose()
|
|
|
|
with pytest.raises(RuntimeError, match="refuses to delete conversation history"):
|
|
await run_migration(db_url)
|
|
|
|
|
|
def test_postgresql_ddl_generation():
|
|
ddl = _conversations_ddl("postgresql")
|
|
assert "TIMESTAMP WITH TIME ZONE" in ddl
|
|
assert "JSONB" in ddl
|
|
assert CONVERSATIONS_TABLE in ddl
|
|
assert "FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs" in ddl
|
|
assert "FOREIGN KEY (internal_session_id) REFERENCES maf_roundtable_sessions" in ddl
|
|
|
|
msg_ddl = _messages_ddl("postgresql")
|
|
assert "ON DELETE CASCADE" in msg_ddl
|
|
assert "UNIQUE (conversation_id, seq)" in msg_ddl
|
|
|
|
evt_ddl = _events_ddl("postgresql")
|
|
assert "UNIQUE (conversation_id, seq)" in evt_ddl
|
|
|
|
|
|
def test_mysql_ddl_generation():
|
|
ddl = _conversations_ddl("mysql")
|
|
assert "DATETIME(6)" in ddl
|
|
assert "JSON" in ddl
|
|
assert "FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs" in ddl
|
|
|
|
msg_ddl = _messages_ddl("mysql")
|
|
assert "LONGTEXT" in msg_ddl
|
|
assert "ON DELETE CASCADE" in msg_ddl
|
|
|
|
|
|
def test_sqlite_ddl_generation():
|
|
ddl = _conversations_ddl("sqlite")
|
|
assert "DATETIME" in ddl
|
|
assert "TEXT" in ddl
|
|
|
|
indexes = _indexes_ddl("sqlite")
|
|
assert len(indexes) >= 8
|
|
assert any("ix_maf_conv_owner_updated" in i for i in indexes)
|
|
assert any("ix_maf_msg_conv_seq" in i for i in indexes)
|
|
assert any("ix_maf_cevt_conv_seq" in i for i in indexes)
|
|
|
|
|
|
# ── In-place upgrade: existing events table missing source_event_id ──────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upgrade_adds_source_event_id_to_existing_events_table(tmp_path):
|
|
"""Old V2 deployments created the events table without source_event_id.
|
|
Running the migration must ALTER TABLE ADD COLUMN + create unique index."""
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'upgrade.db'}"
|
|
|
|
# Phase 1: create tables with the OLD schema (no source_event_id).
|
|
engine_old = create_async_engine(db_url)
|
|
async with engine_old.begin() as conn:
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_sessions ("
|
|
"id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_text TEXT NOT NULL, "
|
|
"task_payload_json TEXT, mode VARCHAR(16) NOT NULL DEFAULT 'structured', "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'draft', model_name VARCHAR(128), "
|
|
"reasoning_effort VARCHAR(32), manager_agent_id VARCHAR(64), "
|
|
"max_rounds INTEGER NOT NULL DEFAULT 8, write_policy VARCHAR(32) NOT NULL DEFAULT 'confirm_high_risk', "
|
|
"capability_snapshot_json TEXT, fact_ledger_json TEXT, final_result_json TEXT, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_runs ("
|
|
"id VARCHAR(64) PRIMARY KEY, session_id VARCHAR(64) NOT NULL, "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'pending', workflow_version VARCHAR(64), "
|
|
"maf_checkpoint_json TEXT, last_event_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"started_at DATETIME, finished_at DATETIME, error_code VARCHAR(64), "
|
|
"error_message_safe TEXT, token_usage_json TEXT, latency_ms INTEGER, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_context_json TEXT, "
|
|
"safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', active_run_id VARCHAR(64), "
|
|
"internal_session_id VARCHAR(64), last_stream_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs (id) ON DELETE SET NULL, "
|
|
"FOREIGN KEY (internal_session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE SET NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {MESSAGES_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, message_kind VARCHAR(32) NOT NULL DEFAULT 'user', "
|
|
"role VARCHAR(16) NOT NULL DEFAULT 'user', agent_id VARCHAR(64), run_id VARCHAR(64), "
|
|
"message_ref VARCHAR(191), content TEXT NOT NULL DEFAULT '', metadata_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_messages_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
# OLD events table: no source_event_id column
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {EVENTS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, run_id VARCHAR(64), "
|
|
"event_type VARCHAR(64) NOT NULL, data_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_conv_events_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
# Insert some existing event data that must be preserved
|
|
await conn.execute(text(
|
|
f"INSERT INTO {CONVERSATIONS_TABLE} (id, owner_user_id, title, safe_status, last_stream_seq, created_at, updated_at) "
|
|
"VALUES ('c1', 'u1', 'OldData', 'idle', 1, '2026-01-01', '2026-01-01')"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {EVENTS_TABLE} (id, conversation_id, seq, run_id, event_type, data_json, created_at) "
|
|
"VALUES ('e1', 'c1', 1, NULL, 'user.message.created', '{}', '2026-01-01')"
|
|
))
|
|
await engine_old.dispose()
|
|
|
|
# Phase 2: run migration — should ALTER TABLE + add unique index
|
|
await run_migration(db_url)
|
|
|
|
# Phase 3: verify the column and index exist, data preserved
|
|
engine_new = create_async_engine(db_url)
|
|
async with engine_new.begin() as conn:
|
|
def _check(sync_conn):
|
|
insp = inspect(sync_conn)
|
|
return validate_schema(insp), insp
|
|
|
|
issues, insp = await conn.run_sync(_check)
|
|
assert issues == [], f"Schema issues after upgrade: {issues}"
|
|
|
|
# Data preserved
|
|
row = (await conn.execute(text(f"SELECT title FROM {CONVERSATIONS_TABLE} WHERE id='c1'"))).first()
|
|
assert row is not None
|
|
assert row[0] == "OldData"
|
|
|
|
evt_row = (await conn.execute(text(f"SELECT event_type FROM {EVENTS_TABLE} WHERE id='e1'"))).first()
|
|
assert evt_row is not None
|
|
assert evt_row[0] == "user.message.created"
|
|
|
|
# source_event_id column exists and is NULL for old rows
|
|
src_row = (await conn.execute(text(f"SELECT source_event_id FROM {EVENTS_TABLE} WHERE id='e1'"))).first()
|
|
assert src_row is not None
|
|
assert src_row[0] is None
|
|
|
|
await engine_new.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upgrade_idempotent_after_alter(tmp_path):
|
|
"""Running migration twice on an upgraded DB must not error."""
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'upgrade2.db'}"
|
|
|
|
engine_old = create_async_engine(db_url)
|
|
async with engine_old.begin() as conn:
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_sessions ("
|
|
"id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_text TEXT NOT NULL, "
|
|
"task_payload_json TEXT, mode VARCHAR(16) NOT NULL DEFAULT 'structured', "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'draft', model_name VARCHAR(128), "
|
|
"reasoning_effort VARCHAR(32), manager_agent_id VARCHAR(64), "
|
|
"max_rounds INTEGER NOT NULL DEFAULT 8, write_policy VARCHAR(32) NOT NULL DEFAULT 'confirm_high_risk', "
|
|
"capability_snapshot_json TEXT, fact_ledger_json TEXT, final_result_json TEXT, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_runs ("
|
|
"id VARCHAR(64) PRIMARY KEY, session_id VARCHAR(64) NOT NULL, "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'pending', workflow_version VARCHAR(64), "
|
|
"maf_checkpoint_json TEXT, last_event_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"started_at DATETIME, finished_at DATETIME, error_code VARCHAR(64), "
|
|
"error_message_safe TEXT, token_usage_json TEXT, latency_ms INTEGER, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_context_json TEXT, "
|
|
"safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', active_run_id VARCHAR(64), "
|
|
"internal_session_id VARCHAR(64), last_stream_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs (id) ON DELETE SET NULL, "
|
|
"FOREIGN KEY (internal_session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE SET NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {MESSAGES_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, message_kind VARCHAR(32) NOT NULL DEFAULT 'user', "
|
|
"role VARCHAR(16) NOT NULL DEFAULT 'user', agent_id VARCHAR(64), run_id VARCHAR(64), "
|
|
"message_ref VARCHAR(191), content TEXT NOT NULL DEFAULT '', metadata_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_messages_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {EVENTS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, run_id VARCHAR(64), "
|
|
"event_type VARCHAR(64) NOT NULL, data_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_conv_events_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await engine_old.dispose()
|
|
|
|
await run_migration(db_url)
|
|
await run_migration(db_url) # Must not error
|
|
|
|
engine_check = create_async_engine(db_url)
|
|
async with engine_check.begin() as conn:
|
|
def _check(sync_conn):
|
|
return validate_schema(inspect(sync_conn))
|
|
assert await conn.run_sync(_check) == []
|
|
await engine_check.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upgrade_preserves_existing_event_data_with_null_source(tmp_path):
|
|
"""Old V2 table with existing event data: upgrade preserves data,
|
|
source_event_id column is added with NULL for historical rows,
|
|
and the unique index is created."""
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'upgrade_data.db'}"
|
|
|
|
engine_old = create_async_engine(db_url)
|
|
async with engine_old.begin() as conn:
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_sessions ("
|
|
"id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_text TEXT NOT NULL, "
|
|
"task_payload_json TEXT, mode VARCHAR(16) NOT NULL DEFAULT 'structured', "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'draft', model_name VARCHAR(128), "
|
|
"reasoning_effort VARCHAR(32), manager_agent_id VARCHAR(64), "
|
|
"max_rounds INTEGER NOT NULL DEFAULT 8, write_policy VARCHAR(32) NOT NULL DEFAULT 'confirm_high_risk', "
|
|
"capability_snapshot_json TEXT, fact_ledger_json TEXT, final_result_json TEXT, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_runs ("
|
|
"id VARCHAR(64) PRIMARY KEY, session_id VARCHAR(64) NOT NULL, "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'pending', workflow_version VARCHAR(64), "
|
|
"maf_checkpoint_json TEXT, last_event_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"started_at DATETIME, finished_at DATETIME, error_code VARCHAR(64), "
|
|
"error_message_safe TEXT, token_usage_json TEXT, latency_ms INTEGER, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_context_json TEXT, "
|
|
"safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', active_run_id VARCHAR(64), "
|
|
"internal_session_id VARCHAR(64), last_stream_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs (id) ON DELETE SET NULL, "
|
|
"FOREIGN KEY (internal_session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE SET NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {MESSAGES_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, message_kind VARCHAR(32) NOT NULL DEFAULT 'user', "
|
|
"role VARCHAR(16) NOT NULL DEFAULT 'user', agent_id VARCHAR(64), run_id VARCHAR(64), "
|
|
"message_ref VARCHAR(191), content TEXT NOT NULL DEFAULT '', metadata_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_messages_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {EVENTS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, run_id VARCHAR(64), "
|
|
"event_type VARCHAR(64) NOT NULL, data_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_conv_events_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {CONVERSATIONS_TABLE} (id, owner_user_id, title, safe_status, last_stream_seq, created_at, updated_at) "
|
|
"VALUES ('c1', 'u1', 'OldData', 'idle', 3, '2026-01-01', '2026-01-01')"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {EVENTS_TABLE} (id, conversation_id, seq, run_id, event_type, data_json, created_at) "
|
|
"VALUES ('e1', 'c1', 1, NULL, 'user.message.created', '{\"content\":\"hello\"}', '2026-01-01')"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {EVENTS_TABLE} (id, conversation_id, seq, run_id, event_type, data_json, created_at) "
|
|
"VALUES ('e2', 'c1', 2, NULL, 'agent.message.delta', '{\"text\":\"hi\"}', '2026-01-01')"
|
|
))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {EVENTS_TABLE} (id, conversation_id, seq, run_id, event_type, data_json, created_at) "
|
|
"VALUES ('e3', 'c1', 3, NULL, 'run.completed', '{\"status\":\"ok\"}', '2026-01-01')"
|
|
))
|
|
await engine_old.dispose()
|
|
|
|
await run_migration(db_url)
|
|
|
|
engine_new = create_async_engine(db_url)
|
|
async with engine_new.begin() as conn:
|
|
def _check(sync_conn):
|
|
insp = inspect(sync_conn)
|
|
return validate_schema(insp)
|
|
|
|
issues = await conn.run_sync(_check)
|
|
assert issues == [], f"Schema issues after upgrade: {issues}"
|
|
|
|
rows = (await conn.execute(text(
|
|
f"SELECT id, event_type, source_event_id FROM {EVENTS_TABLE} ORDER BY seq"
|
|
))).fetchall()
|
|
assert len(rows) == 3
|
|
assert rows[0][0] == "e1"
|
|
assert rows[0][1] == "user.message.created"
|
|
assert rows[0][2] is None
|
|
assert rows[1][0] == "e2"
|
|
assert rows[1][2] is None
|
|
assert rows[2][0] == "e3"
|
|
assert rows[2][2] is None
|
|
|
|
conv_row = (await conn.execute(text(
|
|
f"SELECT title, last_stream_seq FROM {CONVERSATIONS_TABLE} WHERE id='c1'"
|
|
))).first()
|
|
assert conv_row[0] == "OldData"
|
|
assert conv_row[1] == 3
|
|
await engine_new.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_unique_detected_after_upgrade_on_sqlite(tmp_path):
|
|
"""After upgrade on SQLite, _events_table_has_source_unique must return True
|
|
even though the unique was created via CREATE UNIQUE INDEX (not ADD CONSTRAINT).
|
|
This verifies the get_indexes() fallback works correctly."""
|
|
db_url = f"sqlite+aiosqlite:///{tmp_path / 'detect.db'}"
|
|
|
|
engine_old = create_async_engine(db_url)
|
|
async with engine_old.begin() as conn:
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_sessions ("
|
|
"id VARCHAR(64) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_text TEXT NOT NULL, "
|
|
"task_payload_json TEXT, mode VARCHAR(16) NOT NULL DEFAULT 'structured', "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'draft', model_name VARCHAR(128), "
|
|
"reasoning_effort VARCHAR(32), manager_agent_id VARCHAR(64), "
|
|
"max_rounds INTEGER NOT NULL DEFAULT 8, write_policy VARCHAR(32) NOT NULL DEFAULT 'confirm_high_risk', "
|
|
"capability_snapshot_json TEXT, fact_ledger_json TEXT, final_result_json TEXT, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
"CREATE TABLE maf_roundtable_runs ("
|
|
"id VARCHAR(64) PRIMARY KEY, session_id VARCHAR(64) NOT NULL, "
|
|
"status VARCHAR(32) NOT NULL DEFAULT 'pending', workflow_version VARCHAR(64), "
|
|
"maf_checkpoint_json TEXT, last_event_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"started_at DATETIME, finished_at DATETIME, error_code VARCHAR(64), "
|
|
"error_message_safe TEXT, token_usage_json TEXT, latency_ms INTEGER, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {CONVERSATIONS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64) NOT NULL, "
|
|
"title VARCHAR(255) NOT NULL DEFAULT '', task_context_json TEXT, "
|
|
"safe_status VARCHAR(32) NOT NULL DEFAULT 'idle', active_run_id VARCHAR(64), "
|
|
"internal_session_id VARCHAR(64), last_stream_seq INTEGER NOT NULL DEFAULT 0, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, "
|
|
"FOREIGN KEY (active_run_id) REFERENCES maf_roundtable_runs (id) ON DELETE SET NULL, "
|
|
"FOREIGN KEY (internal_session_id) REFERENCES maf_roundtable_sessions (id) ON DELETE SET NULL)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {MESSAGES_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, message_kind VARCHAR(32) NOT NULL DEFAULT 'user', "
|
|
"role VARCHAR(16) NOT NULL DEFAULT 'user', agent_id VARCHAR(64), run_id VARCHAR(64), "
|
|
"message_ref VARCHAR(191), content TEXT NOT NULL DEFAULT '', metadata_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_messages_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {EVENTS_TABLE} ("
|
|
"id VARCHAR(64) PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, "
|
|
"seq INTEGER NOT NULL, run_id VARCHAR(64), "
|
|
"event_type VARCHAR(64) NOT NULL, data_json TEXT, "
|
|
"created_at DATETIME NOT NULL, "
|
|
"CONSTRAINT uq_maf_conv_events_conv_seq UNIQUE (conversation_id, seq), "
|
|
f"FOREIGN KEY (conversation_id) REFERENCES {CONVERSATIONS_TABLE} (id) ON DELETE CASCADE)"
|
|
))
|
|
await engine_old.dispose()
|
|
|
|
await run_migration(db_url)
|
|
|
|
engine_check = create_async_engine(db_url)
|
|
async with engine_check.begin() as conn:
|
|
def _check(sync_conn):
|
|
insp = inspect(sync_conn)
|
|
has_unique = _events_table_has_source_unique(insp)
|
|
issues = validate_schema(insp)
|
|
return has_unique, issues
|
|
|
|
has_unique, issues = await conn.run_sync(_check)
|
|
assert has_unique, "source_event_id unique index not detected on SQLite after upgrade"
|
|
assert issues == [], f"Schema issues: {issues}"
|
|
await engine_check.dispose()
|