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