learningcircuit--local-deep-research/tests/database/test_alembic_migrations.py

"""
Comprehensive tests for Alembic migration functionality.

Tests cover:
- Fresh database migrations
- Existing database handling (pre-Alembic databases)
- Column migrations for existing tables
- Idempotent migrations
- Error scenarios
- SQLCipher encrypted database migrations
"""

import os
from unittest.mock import patch

import pytest
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker

from local_deep_research.database.alembic_runner import (
    get_alembic_config,
    get_current_revision,
    get_head_revision,
    get_migrations_dir,
    needs_migration,
    run_migrations,
    stamp_database,
)
from local_deep_research.database.initialize import initialize_database
from local_deep_research.database.models import Base


class TestAlembicRunner:
    """Tests for the alembic_runner module."""

    @pytest.fixture
    def fresh_engine(self, tmp_path):
        """Create a fresh SQLite engine for testing."""
        db_path = tmp_path / "fresh_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        yield engine
        engine.dispose()

    @pytest.fixture
    def existing_engine(self, tmp_path):
        """Create an engine with existing tables but no Alembic."""
        db_path = tmp_path / "existing_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # Create tables directly (simulating pre-Alembic database)
        Base.metadata.create_all(engine)

        yield engine
        engine.dispose()

    @pytest.fixture
    def partial_engine(self, tmp_path):
        """Create an engine with only some tables (partial initialization)."""
        db_path = tmp_path / "partial_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # Create only a few tables
        from local_deep_research.database.models import Setting, QueueStatus

        Setting.__table__.create(engine, checkfirst=True)
        QueueStatus.__table__.create(engine, checkfirst=True)

        yield engine
        engine.dispose()

    def test_get_migrations_dir_exists(self):
        """Migrations directory should exist."""
        migrations_dir = get_migrations_dir()
        assert migrations_dir.exists()
        assert (migrations_dir / "env.py").exists()
        assert (migrations_dir / "versions").exists()

    def test_get_head_revision_returns_value(self):
        """Should return the head revision."""
        head = get_head_revision()
        assert head is not None
        assert isinstance(head, str)

    def test_get_alembic_config(self, fresh_engine):
        """Should create a valid Alembic config."""
        config = get_alembic_config(fresh_engine)
        assert config is not None
        assert config.get_main_option("script_location") is not None
        assert "engine" not in config.attributes

    def test_fresh_database_has_no_revision(self, fresh_engine):
        """Fresh database should have no current revision."""
        # Fresh database with no tables
        revision = get_current_revision(fresh_engine)
        assert revision is None

    def test_fresh_database_needs_migration(self, fresh_engine):
        """Fresh database should need migration."""
        assert needs_migration(fresh_engine)

    def test_run_migrations_on_fresh_database(self, fresh_engine):
        """Migrations should create all tables on fresh database."""
        run_migrations(fresh_engine)  # raises on failure
        assert not needs_migration(fresh_engine)

        # Verify tables were created
        inspector = inspect(fresh_engine)
        tables = inspector.get_table_names()

        assert "alembic_version" in tables
        assert "settings" in tables
        assert "research" in tables
        assert "task_metadata" in tables

    def test_run_migrations_creates_all_expected_tables(self, fresh_engine):
        """Should create all core tables."""
        run_migrations(fresh_engine)

        inspector = inspect(fresh_engine)
        tables = set(inspector.get_table_names())

        # Core tables that should exist (not exhaustive)
        expected_tables = {
            "settings",
            "research",
            "research_history",
            "journals",
            "app_logs",
            "queued_researches",
            "token_usage",
            "research_ratings",
            "task_metadata",
            "queue_status",
            "alembic_version",
        }

        for table in expected_tables:
            assert table in tables, f"Expected table '{table}' not found"

    def test_existing_database_migration(self, existing_engine):
        """Existing database without Alembic should be migrated properly."""
        # Verify no alembic_version yet
        inspector = inspect(existing_engine)
        tables = inspector.get_table_names()
        assert "alembic_version" not in tables
        assert len(tables) > 0  # Has other tables

        # Run migrations
        run_migrations(existing_engine)  # raises on failure

        # Verify alembic_version now exists
        new_inspector = inspect(existing_engine)
        new_tables = new_inspector.get_table_names()
        assert "alembic_version" in new_tables

        # Verify current revision is set
        revision = get_current_revision(existing_engine)
        assert revision is not None

    def test_partial_database_migration(self, partial_engine):
        """Partial database should have missing tables created."""
        # Verify only some tables exist
        inspector = inspect(partial_engine)
        initial_tables = set(inspector.get_table_names())
        assert "settings" in initial_tables
        assert "research" not in initial_tables

        # Run migrations
        run_migrations(partial_engine)  # raises on failure

        # Verify missing tables were created
        new_inspector = inspect(partial_engine)
        final_tables = set(new_inspector.get_table_names())

        assert "settings" in final_tables  # Still exists
        assert "research" in final_tables  # Now created
        assert "task_metadata" in final_tables  # Now created
        assert "alembic_version" in final_tables

    def test_stamp_database(self, fresh_engine):
        """Should stamp database without running migrations."""
        # Create tables first
        Base.metadata.create_all(fresh_engine)

        # Stamp at head
        stamp_database(fresh_engine, "head")

        # Verify stamped
        revision = get_current_revision(fresh_engine)
        head = get_head_revision()
        assert revision == head

    def test_idempotent_migrations(self, fresh_engine):
        """Running migrations multiple times should be safe."""
        # First run
        run_migrations(fresh_engine)  # raises on failure

        inspector1 = inspect(fresh_engine)
        tables1 = set(inspector1.get_table_names())
        revision1 = get_current_revision(fresh_engine)

        # Second run
        run_migrations(fresh_engine)  # raises on failure

        inspector2 = inspect(fresh_engine)
        tables2 = set(inspector2.get_table_names())
        revision2 = get_current_revision(fresh_engine)

        # Should be identical
        assert tables1 == tables2
        assert revision1 == revision2

        # Third run
        run_migrations(fresh_engine)  # raises on failure

    def test_needs_migration_after_complete(self, fresh_engine):
        """After migrations, needs_migration should return False."""
        run_migrations(fresh_engine)
        assert not needs_migration(fresh_engine)

    def test_run_migrations_skips_upgrade_when_at_head(
        self, fresh_engine, loguru_caplog
    ):
        """When the database is already at head, run_migrations must not
        call command.upgrade() — that would open a no-op write transaction
        and hold a RESERVED lock under isolation_level="IMMEDIATE", blocking
        concurrent readers for no benefit.
        """
        # Migrate to head first
        run_migrations(fresh_engine)
        assert get_current_revision(fresh_engine) == get_head_revision()

        # Second call must short-circuit — command.upgrade() should not run,
        # and the short-circuit log line must be emitted (positive signal
        # that the guard fired, not that the function was gutted).
        #
        # Also pins that the orphan-cleanup and FK toggle introduced by
        # PR #4000 are NOT invoked on the short-circuit path. If the
        # guard is moved BELOW engine.connect() + _disable_fk_for_migration
        # in a future refactor, this test fails — the existing
        # command.upgrade mock alone would not catch that regression.
        with (
            patch(
                "local_deep_research.database.alembic_runner.command.upgrade"
            ) as mock_upgrade,
            patch(
                "local_deep_research.database.alembic_runner."
                "_drop_orphan_alembic_temp_tables"
            ) as mock_drop_orphans,
            patch(
                "local_deep_research.database.alembic_runner."
                "_disable_fk_for_migration"
            ) as mock_disable_fk,
        ):
            with loguru_caplog.at_level("INFO"):
                run_migrations(fresh_engine)
            assert mock_upgrade.call_count == 0
            assert mock_drop_orphans.call_count == 0
            assert mock_disable_fk.call_count == 0
            assert "skipping upgrade" in loguru_caplog.text

    def test_run_migrations_runs_upgrade_on_fresh_db(self, tmp_path):
        """Fresh DB (current revision None) must NOT short-circuit;
        command.upgrade() must run so tables and alembic_version get created.
        """
        db_path = tmp_path / "fresh_for_upgrade.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            assert get_current_revision(engine) is None

            run_migrations(engine)

            # If the short-circuit had applied (it must not), revision
            # would still be None and alembic_version would not exist.
            assert get_current_revision(engine) == get_head_revision()
            assert "alembic_version" in inspect(engine).get_table_names()
        finally:
            engine.dispose()

    def test_imports_without_errors(self):
        """All migration modules import without side effects."""

        env_path = get_migrations_dir() / "env.py"
        assert env_path.exists()


class TestColumnMigrations:
    """Tests for column-level migrations."""

    @pytest.fixture
    def old_schema_engine(self, tmp_path):
        """Create database with old schema (missing progress columns)."""
        db_path = tmp_path / "old_schema.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # Create task_metadata table without progress columns
        # This simulates an old database before those columns were added
        with engine.begin() as conn:
            conn.execute(
                text(
                    """
                CREATE TABLE task_metadata (
                    task_id VARCHAR PRIMARY KEY,
                    status VARCHAR NOT NULL,
                    task_type VARCHAR NOT NULL,
                    created_at DATETIME,
                    started_at DATETIME,
                    completed_at DATETIME,
                    error_message VARCHAR,
                    priority INTEGER DEFAULT 0,
                    retry_count INTEGER DEFAULT 0,
                    max_retries INTEGER DEFAULT 3
                )
            """
                )
            )
            # Create a few more tables to simulate existing database
            conn.execute(
                text(
                    """
                CREATE TABLE settings (
                    id INTEGER PRIMARY KEY,
                    key VARCHAR NOT NULL UNIQUE,
                    value VARCHAR,
                    type VARCHAR DEFAULT 'string'
                )
            """
                )
            )

        yield engine
        engine.dispose()

    def test_old_schema_missing_columns(self, old_schema_engine):
        """Verify old schema is missing expected columns."""
        inspector = inspect(old_schema_engine)
        columns = {
            col["name"] for col in inspector.get_columns("task_metadata")
        }

        assert "task_id" in columns
        assert "status" in columns
        assert "progress_current" not in columns
        assert "progress_total" not in columns
        assert "progress_message" not in columns
        assert "metadata_json" not in columns

    def test_column_migration_adds_missing_columns(self, old_schema_engine):
        """Migration should add missing columns to existing tables."""
        # Run migrations
        run_migrations(old_schema_engine)  # raises on failure

        # Verify columns were added
        inspector = inspect(old_schema_engine)
        columns = {
            col["name"] for col in inspector.get_columns("task_metadata")
        }

        assert "progress_current" in columns
        assert "progress_total" in columns
        assert "progress_message" in columns
        assert "metadata_json" in columns

    def test_column_migration_preserves_data(self, old_schema_engine):
        """Migration should preserve existing data in tables."""
        # Insert test data
        with old_schema_engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type)
                VALUES ('test-123', 'completed', 'research')
            """
                )
            )

        # Run migrations
        run_migrations(old_schema_engine)

        # Verify data preserved
        with old_schema_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT task_id, status FROM task_metadata WHERE task_id = 'test-123'"
                )
            ).fetchone()
            assert result is not None
            assert result[0] == "test-123"
            assert result[1] == "completed"

    def test_column_migration_sets_defaults(self, old_schema_engine):
        """New columns should have correct default values."""
        # Insert test data before migration
        with old_schema_engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type)
                VALUES ('test-456', 'processing', 'benchmark')
            """
                )
            )

        # Run migrations
        run_migrations(old_schema_engine)

        # Check that new columns have defaults
        with old_schema_engine.connect() as conn:
            row = conn.execute(
                text(
                    """
                SELECT progress_current, progress_total, progress_message
                FROM task_metadata WHERE task_id = 'test-456'
            """
                )
            ).fetchone()
            # New columns should have default values or NULL
            # progress_current and progress_total default to 0
            # progress_message is nullable
            assert row is not None
            assert row[0] == 0  # progress_current defaults to 0
            assert row[1] == 0  # progress_total defaults to 0


class TestInitializeDatabaseIntegration:
    """Integration tests for initialize_database with Alembic."""

    @pytest.fixture
    def temp_engine(self, tmp_path):
        """Create a temporary database engine."""
        db_path = tmp_path / "integration_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        yield engine
        engine.dispose()

    def test_initialize_fresh_database(self, temp_engine):
        """initialize_database should work on fresh database."""
        Session = sessionmaker(bind=temp_engine)
        with Session() as session:
            initialize_database(temp_engine, session)

        inspector = inspect(temp_engine)
        tables = inspector.get_table_names()

        assert "alembic_version" in tables
        assert "settings" in tables
        assert len(tables) > 20  # Many tables

    def test_initialize_existing_database(self, temp_engine):
        """initialize_database should work on existing database."""
        # First initialization
        Session = sessionmaker(bind=temp_engine)
        with Session() as session:
            initialize_database(temp_engine, session)

        inspector1 = inspect(temp_engine)
        tables1 = set(inspector1.get_table_names())

        # Second initialization (should be safe)
        with Session() as session:
            initialize_database(temp_engine, session)

        inspector2 = inspect(temp_engine)
        tables2 = set(inspector2.get_table_names())

        assert tables1 == tables2

    def test_initialize_preserves_data(self, temp_engine):
        """initialize_database should preserve existing data."""
        Session = sessionmaker(bind=temp_engine)

        # First initialization
        with Session() as session:
            initialize_database(temp_engine, session)

        # Insert test data into queue_status (simpler schema)
        with temp_engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO queue_status (active_tasks, queued_tasks)
                VALUES (5, 10)
            """
                )
            )

        # Second initialization
        with Session() as session:
            initialize_database(temp_engine, session)

        # Verify data preserved
        with temp_engine.connect() as conn:
            result = conn.execute(
                text("SELECT active_tasks, queued_tasks FROM queue_status")
            ).fetchone()
            assert result is not None
            assert result[0] == 5
            assert result[1] == 10


class TestMigrationEdgeCases:
    """Tests for edge cases and error scenarios."""

    @pytest.fixture
    def temp_engine(self, tmp_path):
        """Create a temporary database engine."""
        db_path = tmp_path / "edge_case_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        yield engine
        engine.dispose()

    def test_migration_with_empty_versions_dir(self, temp_engine, tmp_path):
        """Should handle case when no migrations exist."""
        # This tests the edge case where migrations dir is empty
        # The actual implementation should handle this gracefully
        with patch(
            "local_deep_research.database.alembic_runner.get_head_revision",
            return_value=None,
        ):
            run_migrations(temp_engine)  # raises on failure

    def test_get_current_revision_on_corrupted_table(self, temp_engine):
        """Should handle corrupted alembic_version table."""
        # Create a corrupted alembic_version table
        with temp_engine.begin() as conn:
            conn.execute(
                text("CREATE TABLE alembic_version (version_num VARCHAR)")
            )
            # Don't insert any rows - table exists but is empty

        # Should not crash
        revision = get_current_revision(temp_engine)
        assert revision is None  # Empty table means no revision

    def test_multiple_concurrent_migrations(self, tmp_path):
        """Multiple engines against same DB should work safely."""
        db_path = tmp_path / "concurrent_test.db"

        engine1 = create_engine(f"sqlite:///{db_path}")
        engine2 = create_engine(f"sqlite:///{db_path}")

        try:
            # First migration
            run_migrations(engine1)  # raises on failure

            # Second migration (should be idempotent)
            run_migrations(engine2)  # raises on failure

            # Both should see same revision
            rev1 = get_current_revision(engine1)
            rev2 = get_current_revision(engine2)
            assert rev1 == rev2
        finally:
            engine1.dispose()
            engine2.dispose()


class TestSQLCipherMigrations:
    """Tests for SQLCipher encrypted database migrations."""

    @pytest.fixture
    def sqlcipher_available(self):
        """Check if SQLCipher is available."""
        import importlib.util

        if importlib.util.find_spec("sqlcipher3") is None:
            pytest.skip("SQLCipher not available")
        return True

    @pytest.fixture
    def encrypted_engine(self, tmp_path, sqlcipher_available):
        """Create an encrypted SQLCipher database."""
        import sqlcipher3

        db_path = tmp_path / "encrypted_test.db"
        password = "test_password_123"

        # Create encrypted connection
        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            cursor.close()
            return conn

        engine = create_engine("sqlite://", creator=create_connection)
        yield engine
        engine.dispose()

    def test_encrypted_database_migration(self, encrypted_engine):
        """Should be able to run migrations on encrypted database."""
        run_migrations(encrypted_engine)  # raises on failure

        # Verify tables created
        inspector = inspect(encrypted_engine)
        tables = inspector.get_table_names()
        assert "alembic_version" in tables
        assert "settings" in tables

    def test_encrypted_database_idempotent(self, encrypted_engine):
        """Multiple migrations on encrypted DB should be safe."""
        run_migrations(encrypted_engine)  # raises on failure
        run_migrations(encrypted_engine)  # raises on failure

        rev1 = get_current_revision(encrypted_engine)
        head = get_head_revision()
        assert rev1 == head


class TestAlembicVersionTable:
    """Tests for alembic_version table behavior."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "migrated_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_alembic_version_table_exists(self, migrated_engine):
        """Alembic version table should exist after migration."""
        inspector = inspect(migrated_engine)
        assert "alembic_version" in inspector.get_table_names()

    def test_alembic_version_has_single_row(self, migrated_engine):
        """Alembic version table should have exactly one row."""
        with migrated_engine.connect() as conn:
            result = conn.execute(
                text("SELECT COUNT(*) FROM alembic_version")
            ).fetchone()
            assert result[0] == 1

    def test_alembic_version_matches_head(self, migrated_engine):
        """Alembic version should match head revision."""
        with migrated_engine.connect() as conn:
            result = conn.execute(
                text("SELECT version_num FROM alembic_version")
            ).fetchone()
            assert result[0] == get_head_revision()

    def test_revision_updates_after_migration(self, tmp_path):
        """Revision should update when new migrations run."""
        db_path = tmp_path / "revision_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run migrations
            run_migrations(engine)
            revision = get_current_revision(engine)

            # Verify it's at head
            assert revision == get_head_revision()
        finally:
            engine.dispose()


class TestMigrationChain:
    """Tests for migration chain and ordering."""

    def test_migration_files_exist(self):
        """All expected migration files should exist."""
        versions_dir = get_migrations_dir() / "versions"

        expected_files = [
            "0001_initial_schema.py",
            "0002_add_task_progress_columns.py",
            "0003_add_research_indexes.py",
            "0004_migrate_legacy_app_settings.py",
            "0005_add_resource_document_id.py",
            "0006_journal_quality_system.py",
            "0007_backfill_missing_indexes.py",
            "0008_fix_research_strategy_fk.py",
            "0009_default_fetch_mode_summary.py",
            "0010_add_chat_tables.py",
            "0011_add_collection_is_public.py",
            "0012_add_collection_agent_enabled.py",
            "0013_remove_meta_search_engines.py",
            "0014_benchmark_run_version_and_snapshot.py",
            "0015_drop_document_notes.py",
            "0016_drop_orphaned_cache_tables.py",
        ]

        for filename in expected_files:
            filepath = versions_dir / filename
            assert filepath.exists(), f"Migration file {filename} not found"

    def test_migration_chain_is_valid(self):
        """Migration chain should be properly linked."""
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        migrations_dir = get_migrations_dir()
        config = Config()
        config.set_main_option("script_location", str(migrations_dir))

        script = ScriptDirectory.from_config(config)

        # Get all revisions
        revisions = list(script.walk_revisions())

        # Should have at least 3 revisions
        assert len(revisions) >= 3

        # Head should be the latest migration
        assert script.get_current_head() == get_head_revision()

        # Verify chain link 0010 -> 0009 -> 0008 (the rest of the chain is checked below).
        rev_0010 = script.get_revision("0010")
        assert rev_0010.down_revision == "0009"

        rev_0009 = script.get_revision("0009")
        assert rev_0009.down_revision == "0008"

        rev_0008 = script.get_revision("0008")
        assert rev_0008.down_revision == "0007"

        rev_0007 = script.get_revision("0007")
        assert rev_0007.down_revision == "0006"

        rev_0006 = script.get_revision("0006")
        assert rev_0006.down_revision == "0005"

        rev_0005 = script.get_revision("0005")
        assert rev_0005.down_revision == "0004"

        rev_0004 = script.get_revision("0004")
        assert rev_0004.down_revision == "0003"

        rev_0003 = script.get_revision("0003")
        assert rev_0003.down_revision == "0002"

        rev_0002 = script.get_revision("0002")
        assert rev_0002.down_revision == "0001"

        rev_0001 = script.get_revision("0001")
        assert rev_0001.down_revision is None

    def test_run_to_specific_revision(self, tmp_path):
        """Should be able to run migrations to a specific revision."""
        db_path = tmp_path / "specific_rev_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run only to revision 0001
            run_migrations(engine, target="0001")  # raises on failure

            # Should be at 0001
            revision = get_current_revision(engine)
            assert revision == "0001"

            # Should still need migration (not at head)
            assert needs_migration(engine)

            # Run to head
            run_migrations(engine, target="head")  # raises on failure
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestUsersTableExclusion:
    """Tests to verify users table is correctly excluded from user databases."""

    @pytest.fixture
    def user_db_engine(self, tmp_path):
        """Create a user database (not auth database)."""
        db_path = tmp_path / "user_db_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_users_table_not_in_user_database(self, user_db_engine):
        """Users table should NOT be created in user databases."""
        inspector = inspect(user_db_engine)
        tables = inspector.get_table_names()

        # Users table should NOT exist in user databases
        # It only exists in the auth database
        assert "users" not in tables

    def test_user_specific_tables_exist(self, user_db_engine):
        """User-specific tables should exist."""
        inspector = inspect(user_db_engine)
        tables = set(inspector.get_table_names())

        # These tables should exist in user databases
        user_tables = {
            "settings",
            "research",
            "token_usage",
            "api_keys",
            "task_metadata",
        }

        for table in user_tables:
            assert table in tables, f"User table '{table}' not found"


class TestCheckDatabaseSchemaWithAlembic:
    """Tests for check_database_schema function with Alembic."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "schema_check_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_check_schema_shows_no_missing_tables(self, migrated_engine):
        """After migration, no tables should be missing."""
        from local_deep_research.database.initialize import (
            check_database_schema,
        )

        schema_info = check_database_schema(migrated_engine)

        # Should have many tables
        assert len(schema_info["tables"]) > 20

        # Should have no missing tables (except users which is excluded)
        assert len(schema_info["missing_tables"]) == 0

    def test_check_schema_excludes_users_table(self, migrated_engine):
        """check_database_schema should not report users as missing."""
        from local_deep_research.database.initialize import (
            check_database_schema,
        )

        schema_info = check_database_schema(migrated_engine)

        # Users should not appear in missing tables
        assert "users" not in schema_info["missing_tables"]

        # Users should not appear in existing tables either
        assert "users" not in schema_info["tables"]


class TestNeedsMigrationStates:
    """Tests for needs_migration in various database states."""

    def test_needs_migration_empty_database(self, tmp_path):
        """Empty database should need migration."""
        db_path = tmp_path / "empty_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            assert needs_migration(engine)
        finally:
            engine.dispose()

    def test_needs_migration_partial_tables(self, tmp_path):
        """Database with partial tables should need migration."""
        db_path = tmp_path / "partial_tables_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create only one table
            with engine.begin() as conn:
                conn.execute(
                    text("CREATE TABLE settings (id INTEGER PRIMARY KEY)")
                )

            assert needs_migration(engine)
        finally:
            engine.dispose()

    def test_needs_migration_old_revision(self, tmp_path):
        """Database at old revision should need migration."""
        db_path = tmp_path / "old_revision_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run only to 0001
            run_migrations(engine, target="0001")

            # Should still need migration
            assert needs_migration(engine)
        finally:
            engine.dispose()

    def test_needs_migration_at_head(self, tmp_path):
        """Database at head should not need migration."""
        db_path = tmp_path / "at_head_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            run_migrations(engine)
            assert not needs_migration(engine)
        finally:
            engine.dispose()

    def test_needs_migration_stamped_database(self, tmp_path):
        """Stamped database at head should not need migration."""
        db_path = tmp_path / "stamped_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create tables and stamp
            Base.metadata.create_all(engine)
            stamp_database(engine, "head")

            assert not needs_migration(engine)
        finally:
            engine.dispose()


class TestDataIntegrityAcrossMigrations:
    """Tests to verify data integrity during migrations."""

    @pytest.fixture
    def database_with_data(self, tmp_path):
        """Create a database with existing data."""
        db_path = tmp_path / "data_integrity_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # Run initial migration
        run_migrations(engine, target="0001")

        # Insert test data into multiple tables
        with engine.begin() as conn:
            # Insert into task_metadata
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, priority)
                VALUES
                    ('task-1', 'completed', 'research', 1),
                    ('task-2', 'processing', 'benchmark', 2),
                    ('task-3', 'queued', 'research', 3)
            """
                )
            )

            # Insert into queue_status
            conn.execute(
                text(
                    """
                INSERT INTO queue_status (active_tasks, queued_tasks)
                VALUES (2, 5)
            """
                )
            )

        yield engine
        engine.dispose()

    def test_data_preserved_after_column_migration(self, database_with_data):
        """Data should be preserved when adding columns."""
        # Verify current state
        with database_with_data.connect() as conn:
            count = conn.execute(
                text("SELECT COUNT(*) FROM task_metadata")
            ).fetchone()[0]
            assert count == 3

        # Run migration to add columns
        run_migrations(database_with_data)

        # Verify data still exists
        with database_with_data.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT task_id, status, priority FROM task_metadata ORDER BY task_id"
                )
            ).fetchall()

            assert len(result) == 3
            assert result[0][0] == "task-1"
            assert result[0][1] == "completed"
            assert result[0][2] == 1

    def test_new_columns_have_defaults_for_existing_rows(
        self, database_with_data
    ):
        """New columns should have default values for existing rows."""
        # Run migration
        run_migrations(database_with_data)

        # Check new column values
        with database_with_data.connect() as conn:
            result = conn.execute(
                text(
                    """
                SELECT task_id, progress_current, progress_total
                FROM task_metadata
                ORDER BY task_id
            """
                )
            ).fetchall()

            # All rows should have default values (0) for progress columns
            for row in result:
                # progress_current and progress_total should be 0 or NULL
                assert row[1] is None or row[1] == 0
                assert row[2] is None or row[2] == 0

    def test_multiple_tables_data_preserved(self, database_with_data):
        """Data in multiple tables should be preserved."""
        run_migrations(database_with_data)

        with database_with_data.connect() as conn:
            # Check queue_status
            qs_result = conn.execute(
                text("SELECT active_tasks, queued_tasks FROM queue_status")
            ).fetchone()
            assert qs_result[0] == 2
            assert qs_result[1] == 5

            # Check task_metadata
            tm_count = conn.execute(
                text("SELECT COUNT(*) FROM task_metadata")
            ).fetchone()[0]
            assert tm_count == 3


class TestColumnMigrationIdempotency:
    """Tests for column migration idempotency."""

    @pytest.fixture
    def old_schema_engine(self, tmp_path):
        """Create database with old schema."""
        db_path = tmp_path / "column_idempotent_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        with engine.begin() as conn:
            conn.execute(
                text(
                    """
                CREATE TABLE task_metadata (
                    task_id VARCHAR PRIMARY KEY,
                    status VARCHAR NOT NULL,
                    task_type VARCHAR NOT NULL
                )
            """
                )
            )

        yield engine
        engine.dispose()

    def test_column_migration_idempotent(self, old_schema_engine):
        """Running column migration multiple times should be safe."""
        # First run
        run_migrations(old_schema_engine)

        inspector1 = inspect(old_schema_engine)
        columns1 = {
            col["name"] for col in inspector1.get_columns("task_metadata")
        }

        # Second run (should not fail)
        run_migrations(old_schema_engine)  # raises on failure

        inspector2 = inspect(old_schema_engine)
        columns2 = {
            col["name"] for col in inspector2.get_columns("task_metadata")
        }

        # Columns should be identical
        assert columns1 == columns2

    def test_column_already_exists_no_error(self, tmp_path):
        """Adding column that already exists should not error."""
        db_path = tmp_path / "column_exists_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create table WITH the columns already
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    CREATE TABLE task_metadata (
                        task_id VARCHAR PRIMARY KEY,
                        status VARCHAR NOT NULL,
                        task_type VARCHAR NOT NULL,
                        progress_current INTEGER DEFAULT 0,
                        progress_total INTEGER DEFAULT 0,
                        progress_message VARCHAR,
                        metadata_json TEXT
                    )
                """
                    )
                )

            # Run migrations - should not fail
            run_migrations(engine)  # raises on failure

            # Should be at head
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestMigrationErrorHandling:
    """Tests for migration error handling."""

    def test_migration_raises_on_corrupted_schema(self, tmp_path):
        """Migration should raise when alembic_version has wrong schema."""
        db_path = tmp_path / "error_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create a conflicting table structure
            with engine.begin() as conn:
                # Create alembic_version with wrong schema
                conn.execute(
                    text(
                        "CREATE TABLE alembic_version (bad_column INTEGER PRIMARY KEY)"
                    )
                )
                conn.execute(text("INSERT INTO alembic_version VALUES (999)"))

            with pytest.raises(Exception):
                run_migrations(engine)
        finally:
            engine.dispose()

    def test_migration_with_read_only_database(self, tmp_path):
        """Migration should handle read-only database gracefully."""
        import stat

        db_path = tmp_path / "readonly_test.db"

        # Create and migrate database first
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        engine.dispose()

        # Make database read-only
        os.chmod(db_path, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)

        engine = create_engine(f"sqlite:///{db_path}")
        try:
            # Migration on a read-only DB that's already at head should
            # be a no-op (no schema changes needed), so it must succeed
            # and the DB must remain at head.
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision(), (
                "Read-only DB at head should remain at head after no-op migration"
            )
        finally:
            engine.dispose()
            # Restore permissions for cleanup
            os.chmod(
                db_path,
                stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH,
            )


class TestStampDatabaseBehavior:
    """Tests for stamp_database function."""

    def test_stamp_at_specific_revision(self, tmp_path):
        """Should be able to stamp at a specific revision."""
        db_path = tmp_path / "stamp_specific_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create tables
            Base.metadata.create_all(engine)

            # Stamp at 0001
            stamp_database(engine, "0001")

            # Should be at 0001
            assert get_current_revision(engine) == "0001"

            # Should need migration (not at head)
            assert needs_migration(engine)
        finally:
            engine.dispose()

    def test_stamp_creates_alembic_version_table(self, tmp_path):
        """Stamp should create alembic_version table if missing."""
        db_path = tmp_path / "stamp_creates_table_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create some tables but not alembic_version
            with engine.begin() as conn:
                conn.execute(
                    text("CREATE TABLE settings (id INTEGER PRIMARY KEY)")
                )

            # Verify alembic_version doesn't exist
            inspector = inspect(engine)
            assert "alembic_version" not in inspector.get_table_names()

            # Stamp
            stamp_database(engine, "head")

            # Verify alembic_version now exists
            inspector = inspect(engine)
            assert "alembic_version" in inspector.get_table_names()
        finally:
            engine.dispose()

    def test_stamp_overwrites_existing_revision(self, tmp_path):
        """Stamp should overwrite existing revision."""
        db_path = tmp_path / "stamp_overwrite_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run migrations to 0001
            run_migrations(engine, target="0001")
            assert get_current_revision(engine) == "0001"

            # Stamp at head
            stamp_database(engine, "head")
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestMigrationWithForeignKeys:
    """Tests for migrations with foreign key relationships."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "fk_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_foreign_key_tables_created(self, migrated_engine):
        """Tables with foreign keys should be created correctly."""
        inspector = inspect(migrated_engine)
        tables = inspector.get_table_names()

        # Tables that have foreign key relationships
        fk_tables = [
            "research_history",
            "document_collections",
            "document_chunks",
        ]

        for table in fk_tables:
            assert table in tables, f"FK table '{table}' not found"

    def test_can_insert_with_foreign_keys(self, migrated_engine):
        """Should be able to insert data respecting foreign keys."""
        # Use research_history and research_resources which have a clear FK relationship
        with migrated_engine.begin() as conn:
            # Insert a research history record (uses TEXT columns, UUID-like id)
            conn.execute(
                text(
                    """
                INSERT INTO research_history (id, query, mode, status, created_at)
                VALUES ('test-uuid-001', 'Test Query', 'quick', 'completed', datetime('now'))
            """
                )
            )

            # Insert research resource which references research_history
            conn.execute(
                text(
                    """
                INSERT INTO research_resources (research_id, title, url, created_at)
                VALUES ('test-uuid-001', 'Test Resource', 'http://example.com', datetime('now'))
            """
                )
            )

        # Verify data and foreign key relationship
        with migrated_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT COUNT(*) FROM research_resources WHERE research_id = 'test-uuid-001'"
                )
            ).fetchone()
            assert result[0] == 1


class TestMigrationWithIndexes:
    """Tests for index creation during migrations."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "index_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_indexes_created(self, migrated_engine):
        """Important indexes should be created."""
        inspector = inspect(migrated_engine)

        # Check indexes on settings table
        settings_indexes = inspector.get_indexes("settings")
        index_names = {idx["name"] for idx in settings_indexes}

        # Should have index on key column
        assert any("key" in name.lower() for name in index_names if name)

    def test_primary_keys_exist(self, migrated_engine):
        """Primary keys should exist on all tables."""
        inspector = inspect(migrated_engine)
        tables_checked = 0

        for table_name in inspector.get_table_names():
            if table_name == "alembic_version":
                continue

            pk = inspector.get_pk_constraint(table_name)
            # Most tables should have a primary key
            # (some junction tables might not)
            # Verify we can at least retrieve the constraint
            assert pk is not None
            tables_checked += 1

        # Verify we actually checked some tables
        assert tables_checked > 0


class TestMigrationRobustness:
    """Tests for migration robustness and recovery."""

    def test_migration_after_crash_simulation(self, tmp_path):
        """Database should recover from simulated crash during migration."""
        db_path = tmp_path / "crash_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run partial migration
            run_migrations(engine, target="0001")

            # Simulate "crash" by just disposing without cleanup
            engine.dispose()

            # Reconnect and continue
            engine = create_engine(f"sqlite:///{db_path}")
            run_migrations(engine)  # raises on failure
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_migration_with_wal_mode(self, tmp_path):
        """Migration should work with WAL journal mode."""
        db_path = tmp_path / "wal_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Enable WAL mode
            with engine.begin() as conn:
                conn.execute(text("PRAGMA journal_mode=WAL"))

            # Run migrations
            run_migrations(engine)  # raises on failure

            # Verify
            inspector = inspect(engine)
            assert "alembic_version" in inspector.get_table_names()
        finally:
            engine.dispose()

    def test_migration_with_busy_timeout(self, tmp_path):
        """Migration should respect busy timeout settings."""
        db_path = tmp_path / "busy_test.db"
        engine = create_engine(
            f"sqlite:///{db_path}",
            connect_args={"timeout": 30},
        )

        try:
            run_migrations(engine)
            # Verify migration actually ran to head, not just "didn't raise"
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestAllModelsCreated:
    """Tests to verify all expected models are created."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "all_models_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_research_models_created(self, migrated_engine):
        """Research-related tables should be created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        research_tables = [
            "research",
            "research_history",
            "research_tasks",
            "search_queries",
            "search_results",
        ]

        for table in research_tables:
            assert table in tables, f"Research table '{table}' not found"

    def test_library_models_created(self, migrated_engine):
        """Library-related tables should be created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        library_tables = [
            "collections",
            "documents",
            "document_collections",
            "source_types",
        ]

        for table in library_tables:
            assert table in tables, f"Library table '{table}' not found"

    def test_metrics_models_created(self, migrated_engine):
        """Metrics-related tables should be created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        metrics_tables = [
            "token_usage",
            "model_usage",
            "research_ratings",
        ]

        for table in metrics_tables:
            assert table in tables, f"Metrics table '{table}' not found"

    def test_news_models_created(self, migrated_engine):
        """News-related tables should be created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        news_tables = [
            "news_subscriptions",
            "news_cards",
            "news_interests",
        ]

        for table in news_tables:
            assert table in tables, f"News table '{table}' not found"

    def test_benchmark_models_created(self, migrated_engine):
        """Benchmark-related tables should be created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        benchmark_tables = [
            "benchmark_runs",
            "benchmark_results",
            "benchmark_configs",
        ]

        for table in benchmark_tables:
            assert table in tables, f"Benchmark table '{table}' not found"


class TestColumnTypes:
    """Tests to verify column types are correct after migration."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "column_types_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_task_metadata_columns(self, migrated_engine):
        """task_metadata should have correct column types."""
        inspector = inspect(migrated_engine)
        columns = {
            col["name"]: col for col in inspector.get_columns("task_metadata")
        }

        # Check required columns exist
        assert "task_id" in columns
        assert "status" in columns
        assert "progress_current" in columns
        assert "progress_total" in columns
        assert "progress_message" in columns
        assert "metadata_json" in columns

    def test_settings_columns(self, migrated_engine):
        """settings should have correct column types."""
        inspector = inspect(migrated_engine)
        columns = {
            col["name"]: col for col in inspector.get_columns("settings")
        }

        assert "id" in columns
        assert "key" in columns
        assert "value" in columns
        assert "name" in columns

    def test_datetime_columns_exist(self, migrated_engine):
        """Tables should have datetime columns where expected."""
        inspector = inspect(migrated_engine)

        # Check research table has datetime columns
        research_columns = {
            col["name"] for col in inspector.get_columns("research")
        }
        assert "created_at" in research_columns

        # Check task_metadata has datetime columns
        task_columns = {
            col["name"] for col in inspector.get_columns("task_metadata")
        }
        assert "created_at" in task_columns


class TestMigrationPerformance:
    """Performance-related tests for migrations."""

    def test_migration_completes_in_reasonable_time(self, tmp_path):
        """Migration should complete within reasonable time."""
        import time

        db_path = tmp_path / "perf_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            start = time.time()
            run_migrations(engine)
            elapsed = time.time() - start

            # Should complete in under 30 seconds even on slow systems
            assert elapsed < 30, (
                f"Migration took {elapsed:.1f}s, expected < 30s"
            )
        finally:
            engine.dispose()

    def test_migration_with_large_existing_data(self, tmp_path):
        """Migration should handle tables with existing data."""
        db_path = tmp_path / "large_data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run initial migration
            run_migrations(engine, target="0001")

            # Insert many rows
            with engine.begin() as conn:
                for i in range(100):
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type)
                        VALUES ('task-{i}', 'completed', 'research')
                    """
                        )
                    )

            # Run remaining migrations
            run_migrations(engine)  # raises on failure

            # Verify all data preserved
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 100
        finally:
            engine.dispose()


class TestDatabaseManagerIntegration:
    """Integration tests with DatabaseManager (encrypted databases)."""

    @pytest.fixture
    def sqlcipher_available(self):
        """Check if SQLCipher is available."""
        import importlib.util

        if importlib.util.find_spec("sqlcipher3") is None:
            pytest.skip("SQLCipher not available")
        return True

    def test_migration_through_database_manager_flow(
        self, tmp_path, sqlcipher_available
    ):
        """Test migrations work through the full DatabaseManager flow."""
        import sqlcipher3

        db_path = tmp_path / "dm_flow_test.db"
        password = "test_password"

        # Simulate DatabaseManager creating a new database
        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            cursor.close()
            return conn

        engine = create_engine("sqlite://", creator=create_connection)

        try:
            # This simulates what initialize_database does
            run_migrations(engine)  # raises on failure

            # Verify database is properly set up
            inspector = inspect(engine)
            tables = inspector.get_table_names()

            assert "alembic_version" in tables
            assert "settings" in tables
            assert "task_metadata" in tables

            # Verify we can query
            with engine.connect() as conn:
                result = conn.execute(
                    text("SELECT version_num FROM alembic_version")
                ).fetchone()
                assert result[0] == get_head_revision()
        finally:
            engine.dispose()

    def test_reopening_encrypted_database(self, tmp_path, sqlcipher_available):
        """Test that reopening an encrypted database works correctly."""
        import sqlcipher3

        db_path = tmp_path / "reopen_test.db"
        password = "test_password"

        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            cursor.close()
            return conn

        # First open - create and migrate
        engine1 = create_engine("sqlite://", creator=create_connection)
        run_migrations(engine1)
        rev1 = get_current_revision(engine1)
        engine1.dispose()

        # Second open - should recognize existing migration
        engine2 = create_engine("sqlite://", creator=create_connection)
        run_migrations(engine2)  # raises on failure

        rev2 = get_current_revision(engine2)
        assert rev1 == rev2

        # Should not need migration
        assert not needs_migration(engine2)
        engine2.dispose()


class TestMigrationWithDifferentEngineConfigs:
    """Test migrations with various engine configurations."""

    def test_migration_with_echo_enabled(self, tmp_path):
        """Migration should work with SQL echo enabled."""
        db_path = tmp_path / "echo_test.db"
        engine = create_engine(f"sqlite:///{db_path}", echo=False)

        try:
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_migration_with_pool_pre_ping(self, tmp_path):
        """Migration should work with pool pre-ping enabled."""
        db_path = tmp_path / "preping_test.db"
        engine = create_engine(f"sqlite:///{db_path}", pool_pre_ping=True)

        try:
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_migration_with_static_pool(self, tmp_path):
        """Migration should work with StaticPool."""
        from sqlalchemy.pool import StaticPool

        db_path = tmp_path / "static_pool_test.db"
        engine = create_engine(
            f"sqlite:///{db_path}",
            poolclass=StaticPool,
            connect_args={"check_same_thread": False},
        )

        try:
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_migration_in_memory_database(self):
        """Migration should work with in-memory database."""
        from sqlalchemy.pool import StaticPool

        engine = create_engine(
            "sqlite:///:memory:",
            poolclass=StaticPool,
            connect_args={"check_same_thread": False},
        )

        try:
            run_migrations(engine)  # raises on failure

            inspector = inspect(engine)
            assert "alembic_version" in inspector.get_table_names()
        finally:
            engine.dispose()


class TestDowngradeMigrations:
    """Tests for downgrade functionality."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "downgrade_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_downgrade_from_head_to_specific_revision(self, tmp_path):
        """Test downgrade from head to a specific revision.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        """
        from alembic import command

        db_path = tmp_path / "downgrade_specific_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")
            assert get_current_revision(engine) == "0008"

            # Downgrade to 0001
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Verify revision
            assert get_current_revision(engine) == "0001"

            # Verify progress columns were removed
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" not in columns
            assert "progress_total" not in columns
        finally:
            engine.dispose()

    def test_downgrade_preserves_core_data(self, tmp_path):
        """Test that downgrade preserves data in core columns.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        """
        from alembic import command

        db_path = tmp_path / "downgrade_data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 and insert data
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata (task_id, status, task_type, priority)
                    VALUES ('task-preserve-1', 'completed', 'research', 5)
                """
                    )
                )

            # Downgrade
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Verify core data preserved
            with engine.connect() as conn:
                result = conn.execute(
                    text(
                        "SELECT task_id, status, priority FROM task_metadata WHERE task_id = 'task-preserve-1'"
                    )
                ).fetchone()
                assert result is not None
                assert result[0] == "task-preserve-1"
                assert result[1] == "completed"
                assert result[2] == 5
        finally:
            engine.dispose()

    def test_full_downgrade_to_empty_database(self, tmp_path):
        """Test full downgrade removes all tables.

        Starts from 0008 (last fully-reversible revision) rather than
        head, since 0010 is intentionally non-reversible (SQLite ALTER
        TABLE limitations against legacy unnamed constraints + FK-target
        columns; see NON_REVERSIBLE_REVISIONS).
        """
        from alembic import command

        db_path = tmp_path / "full_downgrade_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible) instead of head
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Verify tables exist
            inspector = inspect(engine)
            tables_before = set(inspector.get_table_names())
            assert len(tables_before) > 10

            # Downgrade to base (empty)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "base")

            # Verify most tables removed (only alembic_version may remain)
            inspector = inspect(engine)
            tables_after = set(inspector.get_table_names())
            # After full downgrade, should have minimal tables
            assert len(tables_after) <= 1  # Only alembic_version or empty
        finally:
            engine.dispose()

    def test_downgrade_then_upgrade_roundtrip(self, tmp_path):
        """Test downgrade and re-upgrade works correctly.

        Starts from 0008 (last fully-reversible revision) rather than
        head, since 0010 is intentionally non-reversible. After the
        roundtrip we re-upgrade to head to verify forward path still
        works through the non-reversible step.
        """
        from alembic import command

        db_path = tmp_path / "roundtrip_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Insert data
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata (task_id, status, task_type)
                    VALUES ('roundtrip-task', 'queued', 'test')
                """
                    )
                )

            # Downgrade to 0001
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Re-upgrade to head (passes through 0010 in forward direction)
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()

            # Verify data still exists
            with engine.connect() as conn:
                result = conn.execute(
                    text(
                        "SELECT task_id FROM task_metadata WHERE task_id = 'roundtrip-task'"
                    )
                ).fetchone()
                assert result is not None
        finally:
            engine.dispose()


class TestMigrationWithExistingData:
    """Tests for migration with various data scenarios."""

    @pytest.fixture
    def engine_at_0001(self, tmp_path):
        """Create a database at revision 0001 for data tests."""
        db_path = tmp_path / "data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine, target="0001")
        yield engine
        engine.dispose()

    def test_migration_with_thousands_of_rows(self, tmp_path):
        """Test migration handles tables with thousands of rows."""
        db_path = tmp_path / "large_data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run initial migration
            run_migrations(engine, target="0001")

            # Insert 5000 rows
            with engine.begin() as conn:
                for batch in range(50):
                    values = ", ".join(
                        [
                            f"('task-{batch}-{i}', 'completed', 'research')"
                            for i in range(100)
                        ]
                    )
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type)
                        VALUES {values}
                    """
                        )
                    )

            # Verify count
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 5000

            # Run remaining migrations
            run_migrations(engine)  # raises on failure

            # Verify all data preserved
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 5000

            # Verify new columns exist
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" in columns
        finally:
            engine.dispose()

    def test_migration_preserves_json_data(self, engine_at_0001):
        """Test migration preserves JSON data in columns."""
        # Insert data with JSON - use research_history which has simpler constraints
        with engine_at_0001.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO research_history (id, query, mode, status, created_at, research_meta)
                VALUES ('json-test-id', 'test query', 'quick', 'completed', datetime('now'),
                        '{"nested": {"key": "value"}, "array": [1, 2, 3]}')
            """
                )
            )

        # Run migration
        run_migrations(engine_at_0001)

        # Verify JSON data preserved
        with engine_at_0001.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT research_meta FROM research_history WHERE id = 'json-test-id'"
                )
            ).fetchone()
            assert result is not None
            import json

            data = json.loads(result[0])
            assert data["nested"]["key"] == "value"
            assert data["array"] == [1, 2, 3]

    def test_migration_preserves_datetime_precision(self, engine_at_0001):
        """Test migration preserves datetime precision."""
        # Insert data with precise datetime
        with engine_at_0001.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, created_at)
                VALUES ('datetime-test', 'completed', 'research', '2025-06-15 14:30:45.123456')
            """
                )
            )

        # Run migration
        run_migrations(engine_at_0001)

        # Verify datetime preserved
        with engine_at_0001.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT created_at FROM task_metadata WHERE task_id = 'datetime-test'"
                )
            ).fetchone()
            assert result is not None
            # SQLite stores as string, verify it contains the microseconds
            datetime_str = str(result[0])
            assert "14:30:45" in datetime_str

    def test_migration_with_null_values_in_nullable_columns(
        self, engine_at_0001
    ):
        """Test migration handles NULL values in nullable columns."""
        # Insert data with NULL values
        with engine_at_0001.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, error_message, started_at, completed_at)
                VALUES ('null-test', 'queued', 'research', NULL, NULL, NULL)
            """
                )
            )

        # Run migration
        run_migrations(engine_at_0001)

        # Verify NULL values preserved and new columns also NULL
        with engine_at_0001.connect() as conn:
            result = conn.execute(
                text(
                    """
                SELECT error_message, started_at, completed_at, progress_message
                FROM task_metadata WHERE task_id = 'null-test'
            """
                )
            ).fetchone()
            assert result is not None
            assert result[0] is None  # error_message
            assert result[1] is None  # started_at
            assert result[2] is None  # completed_at
            assert result[3] is None  # progress_message (new column)

    def test_migration_with_unicode_data(self, engine_at_0001):
        """Test migration preserves unicode and special characters."""
        # Insert unicode data
        with engine_at_0001.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, error_message)
                VALUES ('unicode-test', 'failed', 'research',
                        'Error: 日本語テスト 🔥 Ñoño émojis café')
            """
                )
            )

        # Run migration
        run_migrations(engine_at_0001)

        # Verify unicode preserved
        with engine_at_0001.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT error_message FROM task_metadata WHERE task_id = 'unicode-test'"
                )
            ).fetchone()
            assert result is not None
            assert "日本語テスト" in result[0]
            assert "🔥" in result[0]
            assert "café" in result[0]

    def test_migration_with_empty_strings(self, engine_at_0001):
        """Test migration handles empty string values."""
        with engine_at_0001.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, error_message)
                VALUES ('empty-string-test', 'completed', 'research', '')
            """
                )
            )

        run_migrations(engine_at_0001)

        with engine_at_0001.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT error_message FROM task_metadata WHERE task_id = 'empty-string-test'"
                )
            ).fetchone()
            assert result is not None
            assert result[0] == ""


class TestConcurrentDatabaseAccess:
    """Tests for concurrent database access during migrations."""

    def test_migration_while_database_is_being_read(self, tmp_path):
        """Test migration completes while concurrent reads occur."""
        import threading
        import time

        db_path = tmp_path / "concurrent_read_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine, target="0001")

        # Insert some test data
        with engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type)
                VALUES ('concurrent-test', 'completed', 'research')
            """
                )
            )

        read_results = []
        read_errors = []

        def concurrent_reader():
            """Read from database concurrently."""
            try:
                for _ in range(10):
                    with engine.connect() as conn:
                        result = conn.execute(
                            text("SELECT COUNT(*) FROM task_metadata")
                        ).fetchone()
                        read_results.append(result[0])
                    time.sleep(0.01)
            except Exception as e:
                read_errors.append(str(e))

        try:
            # Start reader thread
            reader = threading.Thread(target=concurrent_reader)
            reader.start()

            # Run migration while reading
            run_migrations(engine)  # raises on failure

            reader.join()

            # Verify reads succeeded
            assert len(read_errors) == 0, f"Read errors: {read_errors}"
            assert len(read_results) > 0
        finally:
            engine.dispose()

    def test_two_engines_migrate_same_database(self, tmp_path):
        """Test two processes trying to migrate same database."""
        db_path = tmp_path / "dual_migrate_test.db"

        engine1 = create_engine(f"sqlite:///{db_path}")
        engine2 = create_engine(f"sqlite:///{db_path}")

        try:
            # Both engines try to migrate
            run_migrations(engine1)  # raises on failure
            run_migrations(engine2)  # raises on failure

            # Both should see same revision
            rev1 = get_current_revision(engine1)
            rev2 = get_current_revision(engine2)
            assert rev1 == rev2 == get_head_revision()
        finally:
            engine1.dispose()
            engine2.dispose()

    def test_migration_with_sqlite_busy_timeout(self, tmp_path):
        """Test migration respects SQLite busy timeout."""
        db_path = tmp_path / "busy_timeout_test.db"

        # Create engine with explicit busy timeout
        engine = create_engine(
            f"sqlite:///{db_path}",
            connect_args={"timeout": 30},
        )

        try:
            run_migrations(engine)  # raises on failure

            # Verify database is correctly migrated
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_sequential_migrations_different_engines(self, tmp_path):
        """Test sequential migrations from different engine instances."""
        db_path = tmp_path / "sequential_test.db"

        # First engine migrates to 0001
        engine1 = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine1, target="0001")
        engine1.dispose()

        # Second engine continues to head
        engine2 = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine2)  # raises on failure
        assert get_current_revision(engine2) == get_head_revision()
        engine2.dispose()

        # Third engine verifies no migration needed
        engine3 = create_engine(f"sqlite:///{db_path}")
        assert not needs_migration(engine3)
        engine3.dispose()


class TestMigrationRecovery:
    """Tests for migration recovery scenarios."""

    def test_recovery_after_partial_migration(self, tmp_path):
        """Test recovery after a partial migration state."""
        db_path = tmp_path / "partial_recovery_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Create a partial state - some tables but not all
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    CREATE TABLE settings (
                        id INTEGER PRIMARY KEY,
                        key VARCHAR UNIQUE,
                        value VARCHAR,
                        name VARCHAR,
                        description VARCHAR,
                        type VARCHAR
                    )
                """
                    )
                )
                conn.execute(
                    text(
                        """
                    CREATE TABLE task_metadata (
                        task_id VARCHAR PRIMARY KEY,
                        status VARCHAR NOT NULL,
                        task_type VARCHAR NOT NULL
                    )
                """
                    )
                )

            # Try to recover by running full migration
            run_migrations(engine)  # raises on failure

            # Verify recovery succeeded
            inspector = inspect(engine)
            tables = inspector.get_table_names()
            assert "alembic_version" in tables
            assert "research" in tables  # Should have been created
        finally:
            engine.dispose()

    def test_database_state_after_interrupted_migration(self, tmp_path):
        """Test database state is consistent after simulated interruption."""
        db_path = tmp_path / "interrupted_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run partial migration
            run_migrations(engine, target="0001")

            # Simulate "interruption" by just stopping here
            engine.dispose()

            # Reconnect and check state
            engine = create_engine(f"sqlite:///{db_path}")

            # Database should be at 0001
            assert get_current_revision(engine) == "0001"

            # Should be able to continue
            run_migrations(engine)  # raises on failure
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_re_running_migration_after_failure(self, tmp_path):
        """Test re-running migration after a previous failure."""
        db_path = tmp_path / "rerun_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # First run succeeds
            run_migrations(engine)  # raises on failure

            # Simulate database being in inconsistent state
            # by manually inserting conflicting data
            with engine.begin() as conn:
                # This shouldn't cause issues on re-run
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata (task_id, status, task_type)
                    VALUES ('conflict-test', 'completed', 'research')
                """
                    )
                )

            # Re-running should still work
            run_migrations(engine)  # raises on failure

            # Data should be preserved
            with engine.connect() as conn:
                result = conn.execute(
                    text(
                        "SELECT task_id FROM task_metadata WHERE task_id = 'conflict-test'"
                    )
                ).fetchone()
                assert result is not None
        finally:
            engine.dispose()

    def test_migration_with_foreign_key_violations_in_source(self, tmp_path):
        """Test migration handles pre-existing data correctly."""
        db_path = tmp_path / "fk_data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run full migration first
            run_migrations(engine)

            # Insert valid data
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata (task_id, status, task_type, priority)
                    VALUES ('fk-test-1', 'completed', 'research', 1),
                           ('fk-test-2', 'queued', 'benchmark', 2)
                """
                    )
                )

            # Re-run migration (should be idempotent)
            run_migrations(engine)  # raises on failure

            # Verify data intact
            with engine.connect() as conn:
                count = conn.execute(
                    text(
                        "SELECT COUNT(*) FROM task_metadata WHERE task_id LIKE 'fk-test%'"
                    )
                ).fetchone()[0]
                assert count == 2
        finally:
            engine.dispose()


class TestSchemaValidation:
    """Tests for schema correctness after migration."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "schema_validation_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_all_foreign_key_relationships_valid(self, migrated_engine):
        """Test that all foreign key relationships reference existing tables."""
        inspector = inspect(migrated_engine)
        existing_tables = set(inspector.get_table_names())

        for table_name in existing_tables:
            if table_name == "alembic_version":
                continue

            fks = inspector.get_foreign_keys(table_name)
            for fk in fks:
                referred_table = fk.get("referred_table")
                if referred_table:
                    assert referred_table in existing_tables, (
                        f"Table {table_name} has FK to non-existent table {referred_table}"
                    )

    def test_all_expected_indexes_exist(self, migrated_engine):
        """Test that important indexes exist."""
        inspector = inspect(migrated_engine)

        # Check settings table has index on key
        if "settings" in inspector.get_table_names():
            indexes = inspector.get_indexes("settings")
            index_columns = []
            for idx in indexes:
                index_columns.extend(idx.get("column_names", []))
            # Settings should have unique constraint or index on key
            # Check if 'key' has unique constraint
            unique_constraints = inspector.get_unique_constraints("settings")
            has_key_constraint = any(
                "key" in uc.get("column_names", []) for uc in unique_constraints
            )
            has_key_index = "key" in index_columns
            assert has_key_constraint or has_key_index, (
                "settings.key should have unique index"
            )

    def test_column_types_match_expectations(self, migrated_engine):
        """Test that column types are as expected."""
        inspector = inspect(migrated_engine)

        # Check task_metadata columns
        tm_columns = {
            col["name"]: col for col in inspector.get_columns("task_metadata")
        }

        # task_id should be string type
        assert (
            "VARCHAR" in str(tm_columns["task_id"]["type"]).upper()
            or "TEXT" in str(tm_columns["task_id"]["type"]).upper()
        )

        # progress_current should be integer
        assert (
            "INTEGER" in str(tm_columns["progress_current"]["type"]).upper()
            or "INT" in str(tm_columns["progress_current"]["type"]).upper()
        )

    def test_nullable_constraints_correct(self, migrated_engine):
        """Test nullable constraints are set correctly."""
        inspector = inspect(migrated_engine)

        # Check task_metadata
        tm_columns = {
            col["name"]: col for col in inspector.get_columns("task_metadata")
        }

        # status should not be nullable
        assert tm_columns["status"]["nullable"] is False

        # error_message should be nullable
        assert tm_columns["error_message"]["nullable"] is True

        # progress_message should be nullable
        assert tm_columns["progress_message"]["nullable"] is True

    def test_primary_keys_defined_correctly(self, migrated_engine):
        """Test all tables have properly defined primary keys."""
        inspector = inspect(migrated_engine)

        for table_name in inspector.get_table_names():
            if table_name == "alembic_version":
                continue

            pk = inspector.get_pk_constraint(table_name)
            pk_columns = pk.get("constrained_columns", [])

            # Most tables should have a primary key
            # (some association tables might not)
            if table_name not in ["document_collections"]:  # Junction tables
                assert len(pk_columns) > 0, (
                    f"Table {table_name} missing primary key"
                )

    def test_all_model_tables_exist(self, migrated_engine):
        """Test that all model tables are created."""
        inspector = inspect(migrated_engine)
        tables = set(inspector.get_table_names())

        # Core tables that must exist
        required_tables = [
            "settings",
            "task_metadata",
            "queue_status",
            "research",
            "research_history",
            "token_usage",
            "benchmark_runs",
            "news_subscriptions",
        ]

        for table in required_tables:
            assert table in tables, f"Required table '{table}' not found"


class TestMigrationVersionConsistency:
    """Tests for migration version tracking consistency."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "version_consistency_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_alembic_version_never_has_multiple_rows(self, migrated_engine):
        """Test that alembic_version table never has multiple rows."""
        with migrated_engine.connect() as conn:
            result = conn.execute(
                text("SELECT COUNT(*) FROM alembic_version")
            ).fetchone()
            assert result[0] == 1, "alembic_version should have exactly one row"

    def test_revision_chain_has_no_gaps(self):
        """Test that the revision chain is continuous with no gaps."""
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        migrations_dir = get_migrations_dir()
        config = Config()
        config.set_main_option("script_location", str(migrations_dir))

        script = ScriptDirectory.from_config(config)

        # Get all revisions
        revisions = list(script.walk_revisions())

        # Build the chain from head to base
        head = script.get_current_head()
        current = head
        chain = []

        while current is not None:
            rev = script.get_revision(current)
            chain.append(current)
            current = rev.down_revision

        # Verify chain matches walk_revisions
        assert len(chain) == len(revisions), "Revision chain has gaps"

    def test_head_revision_matches_latest_migration_file(self):
        """Test that head revision matches the latest migration file."""
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        migrations_dir = get_migrations_dir()
        config = Config()
        config.set_main_option("script_location", str(migrations_dir))

        script = ScriptDirectory.from_config(config)
        head = script.get_current_head()

        # Current head should be the latest migration
        assert head == get_head_revision(), (
            f"Head revision mismatch: expected {get_head_revision()}, got {head}"
        )

    def test_all_revision_files_have_valid_structure(self):
        """Test that all migration files have required functions."""
        versions_dir = get_migrations_dir() / "versions"

        for py_file in versions_dir.glob("*.py"):
            if py_file.name.startswith("__"):
                continue

            content = py_file.read_text()

            # Check required elements
            assert "revision = " in content, f"{py_file.name} missing revision"
            assert "down_revision = " in content, (
                f"{py_file.name} missing down_revision"
            )
            assert "def upgrade(" in content, (
                f"{py_file.name} missing upgrade function"
            )
            assert "def downgrade(" in content, (
                f"{py_file.name} missing downgrade function"
            )

    def test_revision_ids_are_unique(self):
        """Test that all revision IDs are unique."""
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        migrations_dir = get_migrations_dir()
        config = Config()
        config.set_main_option("script_location", str(migrations_dir))

        script = ScriptDirectory.from_config(config)
        revisions = list(script.walk_revisions())

        revision_ids = [rev.revision for rev in revisions]
        assert len(revision_ids) == len(set(revision_ids)), (
            "Duplicate revision IDs found"
        )

    def test_stamp_updates_version_correctly(self, tmp_path):
        """Test that stamping updates version correctly."""
        db_path = tmp_path / "stamp_version_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            Base.metadata.create_all(engine)

            # Stamp at 0001
            stamp_database(engine, "0001")
            assert get_current_revision(engine) == "0001"

            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM alembic_version")
                ).fetchone()[0]
                assert count == 1

            # Stamp at head
            stamp_database(engine, "head")
            assert get_current_revision(engine) == get_head_revision()

            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM alembic_version")
                ).fetchone()[0]
                assert count == 1  # Still only one row
        finally:
            engine.dispose()


class TestEdgeCaseTableNames:
    """Tests for edge cases with table names and SQL handling."""

    @pytest.fixture
    def migrated_engine(self, tmp_path):
        """Create a fully migrated database."""
        db_path = tmp_path / "edge_case_names_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine)
        yield engine
        engine.dispose()

    def test_tables_with_underscores_work(self, migrated_engine):
        """Test that tables with underscores in names work correctly."""
        inspector = inspect(migrated_engine)
        tables = inspector.get_table_names()

        # Find tables with underscores
        underscore_tables = [t for t in tables if "_" in t]
        assert len(underscore_tables) > 0, "Should have tables with underscores"

        # Verify they can be queried
        for table in underscore_tables[:3]:  # Test first 3
            with migrated_engine.connect() as conn:
                # Should not raise
                conn.execute(text(f"SELECT * FROM {table} LIMIT 1"))

    def test_reserved_sql_keywords_handled(self, migrated_engine):
        """Test that any reserved SQL keywords are properly escaped."""
        inspector = inspect(migrated_engine)

        # Check all tables can be queried
        for table_name in inspector.get_table_names():
            with migrated_engine.connect() as conn:
                # This should work even if table name is a reserved word
                conn.execute(text(f'SELECT * FROM "{table_name}" LIMIT 1'))

    def test_column_names_with_special_chars(self, migrated_engine):
        """Test columns can be accessed correctly."""
        inspector = inspect(migrated_engine)

        for table_name in inspector.get_table_names():
            if table_name == "alembic_version":
                continue

            columns = inspector.get_columns(table_name)
            for col in columns[:5]:  # Test first 5 columns per table
                col_name = col["name"]
                with migrated_engine.connect() as conn:
                    # Should be able to select each column
                    conn.execute(
                        text(f'SELECT "{col_name}" FROM "{table_name}" LIMIT 1')
                    )

    def test_case_sensitivity_in_table_names(self, migrated_engine):
        """Test case handling in table names (SQLite is case-insensitive)."""
        inspector = inspect(migrated_engine)
        tables = inspector.get_table_names()

        # SQLite table names are case-insensitive by default
        if "settings" in tables:
            with migrated_engine.connect() as conn:
                # Both should work in SQLite
                conn.execute(text("SELECT * FROM settings LIMIT 1"))
                conn.execute(text("SELECT * FROM SETTINGS LIMIT 1"))

    def test_numeric_prefixed_table_handling(self, migrated_engine):
        """Test handling of tables that might have numeric-looking parts."""
        # Alembic uses revision IDs like 0001, 0002
        # Verify these don't cause issues

        from alembic.config import Config
        from alembic.script import ScriptDirectory

        migrations_dir = get_migrations_dir()
        config = Config()
        config.set_main_option("script_location", str(migrations_dir))

        script = ScriptDirectory.from_config(config)

        # Should be able to get revisions with numeric IDs
        rev_0001 = script.get_revision("0001")
        rev_0002 = script.get_revision("0002")

        assert rev_0001 is not None
        assert rev_0002 is not None

    def test_empty_table_operations(self, migrated_engine):
        """Test operations on empty tables work correctly."""
        inspector = inspect(migrated_engine)

        for table_name in inspector.get_table_names():
            if table_name == "alembic_version":
                continue

            with migrated_engine.connect() as conn:
                # Count on empty table
                result = conn.execute(
                    text(f'SELECT COUNT(*) FROM "{table_name}"')
                ).fetchone()
                # Should return 0 or some number, not error
                assert result[0] >= 0

    def test_long_value_insertion(self, migrated_engine):
        """Test handling of long string values."""
        # Create a very long string
        long_string = "x" * 10000

        with migrated_engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type, error_message)
                VALUES ('long-value-test', 'failed', 'research', :msg)
            """
                ),
                {"msg": long_string},
            )

        # Verify retrieval
        with migrated_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT error_message FROM task_metadata WHERE task_id = 'long-value-test'"
                )
            ).fetchone()
            assert result is not None
            assert len(result[0]) == 10000


class TestSQLCipherMigrationsComprehensive:
    """Comprehensive tests for SQLCipher encrypted database migrations.

    These tests cover scenarios specific to encrypted databases including:
    - Real file-based encrypted databases (not in-memory)
    - Data preservation during migration
    - Wrong password handling
    - Password change followed by migration
    """

    @pytest.fixture
    def sqlcipher_available(self):
        """Check if SQLCipher is available."""
        import importlib.util

        if importlib.util.find_spec("sqlcipher3") is None:
            pytest.skip("SQLCipher not available")
        return True

    @pytest.fixture
    def encrypted_file_engine(self, tmp_path, sqlcipher_available):
        """Create an encrypted file-based SQLCipher database (not in-memory)."""
        import sqlcipher3

        db_path = tmp_path / "encrypted_file_test.db"
        password = "secure_test_password_123!"

        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            cursor.close()
            return conn

        engine = create_engine("sqlite://", creator=create_connection)
        yield {"engine": engine, "db_path": db_path, "password": password}
        engine.dispose()

    def test_migration_on_encrypted_file_database(
        self, encrypted_file_engine, sqlcipher_available
    ):
        """Test migrations work correctly on encrypted file-based databases."""
        engine = encrypted_file_engine["engine"]
        db_path = encrypted_file_engine["db_path"]

        # Run migrations
        run_migrations(engine)  # raises on failure

        # Verify tables created
        inspector = inspect(engine)
        tables = inspector.get_table_names()
        assert "alembic_version" in tables
        assert "settings" in tables
        assert "task_metadata" in tables
        assert "research" in tables

        # Verify the file was actually created (not just in-memory)
        assert db_path.exists()
        # Encrypted file should have some content
        assert db_path.stat().st_size > 0

    def test_migration_preserves_encrypted_data(
        self, tmp_path, sqlcipher_available
    ):
        """Test that data is preserved during migration on encrypted database."""
        import sqlcipher3

        db_path = tmp_path / "data_preserve_encrypted.db"
        password = "data_preserve_password"

        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512")
            cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512")
            cursor.close()
            return conn

        engine = create_engine("sqlite://", creator=create_connection)

        try:
            # Run initial migration to 0001
            run_migrations(engine, target="0001")

            # Insert test data
            with engine.begin() as conn:
                for i in range(100):
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type, priority)
                        VALUES ('encrypted-task-{i}', 'completed', 'research', {i})
                    """
                        )
                    )

            # Verify data count before migration
            with engine.connect() as conn:
                count_before = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count_before == 100

            # Run remaining migrations
            run_migrations(engine)  # raises on failure

            # Verify all data preserved after migration
            with engine.connect() as conn:
                count_after = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count_after == 100

                # Verify specific data integrity
                sample = conn.execute(
                    text(
                        "SELECT task_id, priority FROM task_metadata WHERE task_id = 'encrypted-task-50'"
                    )
                ).fetchone()
                assert sample is not None
                assert sample[1] == 50
        finally:
            engine.dispose()

    def test_wrong_password_fails_gracefully(
        self, tmp_path, sqlcipher_available
    ):
        """Test that using wrong password on encrypted database fails gracefully."""
        import sqlcipher3

        db_path = tmp_path / "wrong_password_test.db"
        correct_password = "correct_password_123"
        wrong_password = "wrong_password_456"

        # Create database with correct password
        def create_connection_correct():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{correct_password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.close()
            return conn

        engine1 = create_engine("sqlite://", creator=create_connection_correct)
        run_migrations(engine1)
        engine1.dispose()

        # Try to open with wrong password
        def create_connection_wrong():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{wrong_password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.close()
            return conn

        engine2 = create_engine("sqlite://", creator=create_connection_wrong)

        # Trying to access the database should fail
        with pytest.raises(Exception):
            with engine2.connect() as conn:
                conn.execute(text("SELECT * FROM settings LIMIT 1"))

        engine2.dispose()

    def test_encrypted_database_reopen_preserves_revision(
        self, tmp_path, sqlcipher_available
    ):
        """Test that reopening encrypted database preserves migration revision."""
        import sqlcipher3

        db_path = tmp_path / "reopen_revision_test.db"
        password = "reopen_test_password"

        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.close()
            return conn

        # First session - create and migrate
        engine1 = create_engine("sqlite://", creator=create_connection)
        run_migrations(engine1)
        revision1 = get_current_revision(engine1)
        engine1.dispose()

        # Second session - reopen and check
        engine2 = create_engine("sqlite://", creator=create_connection)
        revision2 = get_current_revision(engine2)

        assert revision1 == revision2
        assert revision2 == get_head_revision()

        # Should not need migration
        assert not needs_migration(engine2)
        engine2.dispose()

    def test_encrypted_migration_with_large_data(
        self, tmp_path, sqlcipher_available
    ):
        """Test migration on encrypted database with realistic data volume."""
        import sqlcipher3

        db_path = tmp_path / "large_encrypted_test.db"
        password = "large_data_password"

        def create_connection():
            conn = sqlcipher3.connect(str(db_path), check_same_thread=False)
            cursor = conn.cursor()
            cursor.execute(f"PRAGMA key = '{password}'")
            cursor.execute("PRAGMA cipher_page_size = 4096")
            cursor.execute("PRAGMA kdf_iter = 256000")
            cursor.close()
            return conn

        engine = create_engine("sqlite://", creator=create_connection)

        try:
            # Run initial migration
            run_migrations(engine, target="0001")

            # Insert 1000 rows (realistic for a user database)
            with engine.begin() as conn:
                for batch in range(10):
                    values = ", ".join(
                        [
                            f"('large-task-{batch}-{i}', 'completed', 'research')"
                            for i in range(100)
                        ]
                    )
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type)
                        VALUES {values}
                    """
                        )
                    )

            # Run column migration (adds columns to existing table)
            run_migrations(engine)  # raises on failure

            # Verify data preserved
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 1000

            # Verify new columns exist and have defaults
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" in columns
            assert "progress_total" in columns
        finally:
            engine.dispose()


class TestMigrationPerformanceBenchmarks:
    """Performance benchmark tests for migrations.

    These tests establish baselines for migration performance to catch
    any regressions in batch mode operations on large tables.
    """

    @pytest.mark.slow
    def test_batch_migration_performance_small_table(self, tmp_path):
        """Batch column migration should complete quickly on small tables (1K rows)."""
        import time

        db_path = tmp_path / "perf_small_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run initial migration
            run_migrations(engine, target="0001")

            # Insert 1,000 rows
            with engine.begin() as conn:
                for batch in range(10):
                    values = ", ".join(
                        [
                            f"('small-perf-{batch}-{i}', 'completed', 'research')"
                            for i in range(100)
                        ]
                    )
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type)
                        VALUES {values}
                    """
                        )
                    )

            # Time the batch column migration
            start = time.time()
            run_migrations(engine)  # raises on failure
            elapsed = time.time() - start

            # 1K rows should migrate in under 5 seconds
            assert elapsed < 5, (
                f"Small table migration took {elapsed:.2f}s, expected < 5s"
            )

            # Verify data preserved
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 1000
        finally:
            engine.dispose()

    @pytest.mark.slow
    def test_batch_migration_performance_medium_table(self, tmp_path):
        """Batch column migration should complete reasonably on medium tables (10K rows)."""
        import time

        db_path = tmp_path / "perf_medium_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run initial migration
            run_migrations(engine, target="0001")

            # Insert 10,000 rows
            with engine.begin() as conn:
                for batch in range(100):
                    values = ", ".join(
                        [
                            f"('medium-perf-{batch}-{i}', 'completed', 'research')"
                            for i in range(100)
                        ]
                    )
                    conn.execute(
                        text(
                            f"""
                        INSERT INTO task_metadata (task_id, status, task_type)
                        VALUES {values}
                    """
                        )
                    )

            # Verify row count
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 10000

            # Time the batch column migration
            start = time.time()
            run_migrations(engine)  # raises on failure
            elapsed = time.time() - start

            # 10K rows should migrate in under 30 seconds
            assert elapsed < 30, (
                f"Medium table migration took {elapsed:.2f}s, expected < 30s"
            )

            # Verify data preserved
            with engine.connect() as conn:
                count = conn.execute(
                    text("SELECT COUNT(*) FROM task_metadata")
                ).fetchone()[0]
                assert count == 10000

            # Verify new columns have correct defaults
            with engine.connect() as conn:
                sample = conn.execute(
                    text(
                        """
                    SELECT progress_current, progress_total
                    FROM task_metadata LIMIT 1
                """
                    )
                ).fetchone()
                # New columns should have defaults
                assert sample[0] == 0 or sample[0] is None
                assert sample[1] == 0 or sample[1] is None
        finally:
            engine.dispose()

    def test_initial_migration_performance(self, tmp_path):
        """Initial schema creation should be fast."""
        import time

        db_path = tmp_path / "initial_perf_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            start = time.time()
            run_migrations(engine)  # raises on failure
            elapsed = time.time() - start

            # Fresh migration should complete in under 10 seconds
            assert elapsed < 10, (
                f"Initial migration took {elapsed:.2f}s, expected < 10s"
            )

            # Verify all tables created
            inspector = inspect(engine)
            tables = inspector.get_table_names()
            assert len(tables) > 20, "Should create many tables"
        finally:
            engine.dispose()


class TestConcurrentMigrationProcesses:
    """Tests for concurrent migration scenarios.

    Tests what happens when multiple processes/threads try to migrate
    the same database. Note: SQLite has limited concurrency support,
    so we test with staggered timing and sequential engine usage.
    """

    def test_sequential_engine_migrations(self, tmp_path):
        """Test sequential migrations from different engine instances.

        This tests the realistic scenario where different processes/sessions
        try to migrate the same database at different times. SQLite uses
        file-level locking which means truly parallel migrations would
        cause lock contention.
        """
        db_path = tmp_path / "sequential_engine_test.db"

        for i in range(5):
            engine = create_engine(f"sqlite:///{db_path}")
            try:
                run_migrations(engine)  # raises on failure
            finally:
                engine.dispose()

        # Verify database is correctly migrated
        engine = create_engine(f"sqlite:///{db_path}")
        try:
            assert get_current_revision(engine) == get_head_revision()
            assert not needs_migration(engine)
        finally:
            engine.dispose()

    def test_migration_during_active_reads(self, tmp_path):
        """Test migration completes correctly while reads are happening."""
        import threading
        import time

        db_path = tmp_path / "active_reads_test.db"

        # Create and migrate database first
        engine = create_engine(f"sqlite:///{db_path}")
        run_migrations(engine, target="0001")

        # Insert data
        with engine.begin() as conn:
            conn.execute(
                text(
                    """
                INSERT INTO task_metadata (task_id, status, task_type)
                VALUES ('read-test', 'completed', 'research')
            """
                )
            )

        read_counts = []
        read_errors = []
        migration_done = threading.Event()

        def reader_thread():
            """Continuously read from database."""
            read_engine = create_engine(f"sqlite:///{db_path}")
            try:
                while not migration_done.is_set():
                    with read_engine.connect() as conn:
                        result = conn.execute(
                            text("SELECT COUNT(*) FROM task_metadata")
                        ).fetchone()
                        read_counts.append(result[0])
                    time.sleep(0.01)
            except Exception as e:
                read_errors.append(str(e))
            finally:
                read_engine.dispose()

        # Start reader thread
        reader = threading.Thread(target=reader_thread)
        reader.start()

        # Run migration while reads are happening
        time.sleep(0.05)  # Let some reads happen first
        run_migrations(engine)  # raises on failure
        migration_done.set()
        reader.join()
        engine.dispose()

        # Verify results
        assert len(read_errors) == 0, f"Read errors: {read_errors}"
        assert len(read_counts) > 0, "Should have completed some reads"

    def test_no_corruption_on_sequential_migrations(self, tmp_path):
        """Test that sequential migrations from different engines don't corrupt the database."""
        db_path = tmp_path / "no_corruption_test.db"

        # Run migrations sequentially from different engine instances
        for i in range(3):
            engine = create_engine(f"sqlite:///{db_path}")
            try:
                run_migrations(engine)  # raises on failure
            finally:
                engine.dispose()

        # Verify integrity
        check_engine = create_engine(f"sqlite:///{db_path}")
        try:
            with check_engine.connect() as conn:
                # Run quick integrity check
                result = conn.execute(text("PRAGMA quick_check")).fetchone()
                assert result[0] == "ok"

                # Verify alembic_version has exactly one row
                version_count = conn.execute(
                    text("SELECT COUNT(*) FROM alembic_version")
                ).fetchone()[0]
                assert version_count == 1
        finally:
            check_engine.dispose()


class TestDowngradeVerification:
    """Tests to verify and document downgrade behavior.

    These tests document the expected behavior of downgrade migrations,
    including destructive operations.
    """

    def test_downgrade_0002_removes_progress_columns(self, tmp_path):
        """Test that downgrading from 0002 removes the progress columns.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        """
        from alembic import command

        db_path = tmp_path / "downgrade_0002_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Verify progress columns exist
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" in columns
            assert "progress_total" in columns
            assert "progress_message" in columns
            assert "metadata_json" in columns

            # Downgrade to 0001
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Verify progress columns removed
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" not in columns
            assert "progress_total" not in columns
            assert "progress_message" not in columns
            assert "metadata_json" not in columns

            # Core columns should still exist
            assert "task_id" in columns
            assert "status" in columns
        finally:
            engine.dispose()

    def test_downgrade_0001_is_destructive_warning(self, tmp_path):
        """WARNING: Test documents that downgrade from 0001 to base drops ALL tables.

        This test verifies the destructive nature of the 0001 downgrade.
        In production, this should NEVER be run without a backup.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        """
        from alembic import command

        db_path = tmp_path / "destructive_downgrade_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Count tables before downgrade
            inspector = inspect(engine)
            tables_before = set(inspector.get_table_names())
            assert len(tables_before) > 10, "Should have many tables"
            assert "settings" in tables_before
            assert "research" in tables_before

            # Downgrade to base (DESTRUCTIVE)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "base")

            # Verify most tables dropped
            inspector = inspect(engine)
            tables_after = set(inspector.get_table_names())
            # Only alembic_version may remain (or nothing)
            assert len(tables_after) <= 1, (
                f"Expected most tables dropped, got {tables_after}"
            )
        finally:
            engine.dispose()

    def test_downgrade_preserves_data_in_kept_columns(self, tmp_path):
        """Test that downgrade preserves data in columns that are kept.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        """
        from alembic import command

        db_path = tmp_path / "downgrade_preserve_data_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Insert data using both old and new columns
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata
                    (task_id, status, task_type, priority, progress_current, progress_total)
                    VALUES ('downgrade-data-test', 'completed', 'research', 42, 10, 100)
                """
                    )
                )

            # Downgrade to 0001 (removes progress columns but keeps core columns)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Verify core data preserved
            with engine.connect() as conn:
                result = conn.execute(
                    text(
                        """
                    SELECT task_id, status, task_type, priority
                    FROM task_metadata WHERE task_id = 'downgrade-data-test'
                """
                    )
                ).fetchone()
                assert result is not None
                assert result[0] == "downgrade-data-test"
                assert result[1] == "completed"
                assert result[2] == "research"
                assert result[3] == 42  # Priority preserved
        finally:
            engine.dispose()

    def test_upgrade_after_downgrade_restores_columns(self, tmp_path):
        """Test that upgrading after downgrade correctly restores columns.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        Re-upgrade to head still verifies forward path through 0010.
        """
        from alembic import command

        db_path = tmp_path / "upgrade_after_downgrade_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(engine)
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Insert data
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                    INSERT INTO task_metadata (task_id, status, task_type)
                    VALUES ('roundtrip-test', 'completed', 'research')
                """
                    )
                )

            # Downgrade to 0001
            with engine.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # Upgrade back to head (passes through 0010 in forward direction)
            run_migrations(engine)

            # Verify columns restored
            inspector = inspect(engine)
            columns = {
                col["name"] for col in inspector.get_columns("task_metadata")
            }
            assert "progress_current" in columns
            assert "progress_total" in columns

            # Verify data preserved
            with engine.connect() as conn:
                result = conn.execute(
                    text(
                        "SELECT task_id, status FROM task_metadata WHERE task_id = 'roundtrip-test'"
                    )
                ).fetchone()
                assert result is not None
                assert result[0] == "roundtrip-test"
        finally:
            engine.dispose()


# =============================================================================
# Security Tests
# =============================================================================


class TestDatabaseFileSecurity:
    """Tests for database file security properties."""

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_database_file_not_world_readable(self, tmp_path):
        """Database files should not be world-readable after creation."""
        db_path = tmp_path / "permission_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            run_migrations(engine)

            # Check file permissions - verify file exists and is readable
            # Note: SQLite doesn't set restrictive permissions by default,
            # but this test documents the current behavior and verifies umask is applied
            assert db_path.exists(), "Database file should exist"
            file_mode = os.stat(db_path).st_mode
            # The file should have been created with current umask
            # This assertion documents that permissions are set
            assert file_mode & 0o600, "Database should be readable by owner"
        finally:
            engine.dispose()

    def test_encrypted_db_unreadable_without_password(self, tmp_path):
        """Verify encrypted DB cannot be accessed without correct password."""
        pytest.importorskip("sqlcipher3")

        from local_deep_research.database.sqlcipher_utils import (
            create_sqlcipher_connection,
            set_sqlcipher_key,
            apply_sqlcipher_pragmas,
        )
        from local_deep_research.database.sqlcipher_compat import (
            get_sqlcipher_module,
        )

        db_path = tmp_path / "encrypted_test.db"
        password = "correct_password_123"

        # Create encrypted database
        conn = create_sqlcipher_connection(str(db_path), password)
        cursor = conn.cursor()
        cursor.execute(
            "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)"
        )
        cursor.execute("INSERT INTO test_data (value) VALUES ('secret')")
        conn.commit()
        conn.close()

        # Try to open with wrong password
        sqlcipher3 = get_sqlcipher_module()
        wrong_conn = sqlcipher3.connect(str(db_path))
        wrong_cursor = wrong_conn.cursor()
        set_sqlcipher_key(wrong_cursor, "wrong_password")
        apply_sqlcipher_pragmas(wrong_cursor, creation_mode=False)

        with pytest.raises(Exception):
            # Should fail to read data with wrong password
            wrong_cursor.execute("SELECT * FROM test_data")
            wrong_cursor.fetchall()

        wrong_conn.close()

    def test_encrypted_db_readable_with_correct_password(self, tmp_path):
        """Verify encrypted DB can be accessed with correct password."""
        pytest.importorskip("sqlcipher3")

        from local_deep_research.database.sqlcipher_utils import (
            create_sqlcipher_connection,
        )

        db_path = tmp_path / "encrypted_correct_test.db"
        password = "test_password_456"

        # Create encrypted database with test data
        conn = create_sqlcipher_connection(str(db_path), password)
        cursor = conn.cursor()
        cursor.execute(
            "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)"
        )
        cursor.execute("INSERT INTO test_data (value) VALUES ('accessible')")
        conn.commit()
        conn.close()

        # Reopen with correct password
        conn2 = create_sqlcipher_connection(str(db_path), password)
        cursor2 = conn2.cursor()
        cursor2.execute("SELECT value FROM test_data")
        result = cursor2.fetchone()
        conn2.close()

        assert result is not None
        assert result[0] == "accessible"


class TestErrorMessageSanitization:
    """Tests for error message sanitization to prevent information leakage."""

    def test_migration_error_does_not_expose_full_path(self, tmp_path):
        """Error messages should not expose full filesystem paths."""
        # Test with a path that would be sensitive if exposed
        sensitive_path = tmp_path / "sensitive_user_data" / "private.db"
        sensitive_path.parent.mkdir(parents=True, exist_ok=True)

        # Create a read-only directory scenario
        engine = create_engine(f"sqlite:///{sensitive_path}")

        try:
            # This test verifies that if errors occur, they don't leak paths
            # in a way that could be exploited
            run_migrations(engine)

            # If we get here, migration worked - verify database exists
            assert sensitive_path.exists()
        finally:
            engine.dispose()

    def test_invalid_revision_error_is_sanitized(self, tmp_path):
        """Invalid revision errors should not expose internal details."""
        db_path = tmp_path / "revision_error_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # First run valid migrations
            run_migrations(engine)

            # Try to migrate to a non-existent revision
            # This should fail gracefully without exposing internals
            try:
                run_migrations(engine, target="non_existent_revision_xyz")
            except Exception as e:
                error_msg = str(e)
                # Error message should not contain sensitive paths or details
                # that could be used for reconnaissance
                assert "/home/" not in error_msg or "non_existent" in error_msg
        finally:
            engine.dispose()

    def test_connection_error_no_credential_exposure(self):
        """Connection errors should not expose credentials in messages."""
        # Create an engine with invalid credentials in URL
        # (SQLite doesn't use credentials, but the pattern is important)

        # Test that error handling doesn't echo back sensitive info
        try:
            # This should fail but not expose the path in a dangerous way
            engine = create_engine("sqlite:///nonexistent_path/db.sqlite")
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))
        except Exception as e:
            error_msg = str(e).lower()
            # Should not contain common credential patterns
            assert "password" not in error_msg
            assert "secret" not in error_msg
            assert "key=" not in error_msg

    def test_alembic_runner_logs_sanitized(self, tmp_path, caplog):
        """Verify that alembic runner logs don't expose sensitive data."""
        db_path = tmp_path / "log_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # DEBUG level value is 10 in the standard hierarchy
            with caplog.at_level(10):
                run_migrations(engine)

            # Check log messages don't contain sensitive patterns
            for record in caplog.records:
                msg = record.getMessage().lower()
                # Should not log passwords or keys
                assert (
                    "password" not in msg or "wrong" in msg
                )  # test messages ok
                assert "secret" not in msg
                assert "credential" not in msg
        finally:
            engine.dispose()


class TestCryptographicConsistency:
    """Tests for cryptographic consistency in SQLCipher operations."""

    def test_rekey_uses_pbkdf2_like_set_key(self, tmp_path):
        """Verify rekey uses same PBKDF2 derivation as set_key for consistency."""
        pytest.importorskip("sqlcipher3")

        from local_deep_research.database.sqlcipher_utils import (
            create_sqlcipher_connection,
            set_sqlcipher_rekey,
            set_sqlcipher_key,
            apply_sqlcipher_pragmas,
        )
        from local_deep_research.database.sqlcipher_compat import (
            get_sqlcipher_module,
        )

        db_path = tmp_path / "rekey_consistency_test.db"
        original_password = "original_password_123"
        new_password = "new_password_456"

        # Create encrypted database with original password
        conn = create_sqlcipher_connection(str(db_path), original_password)
        cursor = conn.cursor()
        cursor.execute(
            "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)"
        )
        cursor.execute("INSERT INTO test_data (value) VALUES ('secret_data')")
        conn.commit()

        # Rekey to new password
        set_sqlcipher_rekey(cursor, new_password)
        conn.commit()
        conn.close()

        # Verify we can open with new password using set_sqlcipher_key
        # This proves rekey used the same PBKDF2 derivation
        sqlcipher3 = get_sqlcipher_module()
        conn2 = sqlcipher3.connect(str(db_path))
        cursor2 = conn2.cursor()
        set_sqlcipher_key(cursor2, new_password)
        apply_sqlcipher_pragmas(cursor2, creation_mode=False)

        # Should be able to read data with new password
        cursor2.execute("SELECT value FROM test_data")
        result = cursor2.fetchone()
        conn2.close()

        assert result is not None
        assert result[0] == "secret_data"

    def test_rekey_derived_key_matches_set_key_derived_key(self):
        """Verify _get_key_from_password produces consistent results."""
        from local_deep_research.database.sqlcipher_utils import (
            _get_key_from_password,
        )

        password = "test_password_xyz"
        salt = os.urandom(16)
        kdf_iterations = 256000

        # Call twice to ensure consistency (cached result should match)
        key1 = _get_key_from_password(password, salt, kdf_iterations)
        key2 = _get_key_from_password(password, salt, kdf_iterations)

        assert key1 == key2
        assert len(key1) > 0
        # Verify it's a bytes object (PBKDF2 output)
        assert isinstance(key1, bytes)


def _insert_witness_then_fail(config, target):
    """side_effect for patched ``command.upgrade``.

    Inserts a witness row through the transaction's connection, then
    raises. If ``engine.begin()``'s rollback works, the row must not
    be present after the raise — this is what makes the rollback
    tests non-tautological. A plain ``side_effect=RuntimeError(...)``
    fires before any DB write, so the post-failure "revision
    preserved" assertion would be trivially true.

    Uses DML (INSERT) rather than DDL (CREATE TABLE) because pysqlite
    does not reliably roll back DDL statements on default SQLite. The
    caller must pre-create a ``rollback_witness`` table before
    patching ``command.upgrade`` with this side_effect.
    """
    conn = config.attributes["connection"]
    conn.execute(text("INSERT INTO rollback_witness (x) VALUES (1)"))
    raise RuntimeError("simulated failure")


class TestMigrationErrorSanitization:
    """Tests for migration error message sanitization."""

    def test_migration_error_logs_type_not_full_path(self, tmp_path, caplog):
        """Verify migration errors log exception type, not full paths."""
        # Create a scenario that will cause a migration error
        db_path = tmp_path / "error_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Run migrations first
            run_migrations(engine)

            # Now corrupt the alembic_version to cause an error
            with engine.begin() as conn:
                conn.execute(text("DELETE FROM alembic_version"))
                conn.execute(
                    text(
                        "INSERT INTO alembic_version (version_num) VALUES ('invalid')"
                    )
                )

            # Clear logs and try to migrate again
            caplog.clear()

            # This should raise due to the invalid version
            with caplog.at_level(10):  # DEBUG level
                with pytest.raises(Exception):
                    run_migrations(engine)

            # Check that any error logs don't contain sensitive home paths
            for record in caplog.records:
                msg = record.getMessage()
                # Should not contain home directory paths in error messages
                if "error" in msg.lower() or "failed" in msg.lower():
                    # Error messages should be sanitized
                    assert "/home/" not in msg or "Migration" in msg
        finally:
            engine.dispose()

    def test_migration_raises_on_failure_and_preserves_revision(self, tmp_path):
        """Failed migration must raise, leave DB at previous revision,
        AND roll back any writes made inside the transaction.

        Uses a fresh DB so the short-circuit for already-at-head does not
        bypass the patched failure. A fresh DB has current revision None,
        which falls through to command.upgrade().

        The patched side_effect inserts a witness row before raising,
        so the rollback assertion is a real rollback proof — not a
        tautology.
        """
        db_path = tmp_path / "failure_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Pre-create the witness table outside the transaction the
            # patched call will run in. DML (INSERT) rollback is
            # reliable on default SQLite; DDL (CREATE TABLE) is not.
            with engine.begin() as conn:
                conn.execute(text("CREATE TABLE rollback_witness (x INTEGER)"))

            rev_before = get_current_revision(engine)
            assert rev_before is None

            with patch(
                "local_deep_research.database.alembic_runner.command.upgrade",
                side_effect=_insert_witness_then_fail,
            ):
                with pytest.raises(RuntimeError, match="simulated failure"):
                    run_migrations(engine)

            assert get_current_revision(engine) == rev_before
            with engine.begin() as conn:
                rows = list(
                    conn.execute(text("SELECT x FROM rollback_witness"))
                )
            assert rows == []
        finally:
            engine.dispose()

    def test_migration_failure_preserves_concrete_prior_revision(
        self, tmp_path
    ):
        """Failed upgrade from a concrete prior revision must roll back
        cleanly — the DB must still report that concrete revision, and
        any write made inside the transaction must be undone.

        Passes an explicit revision target (not "head") so the at-head
        short-circuit does not skip the patched upgrade call. The
        patched side_effect inserts a witness row before raising, so
        the rollback assertion actually exercises rollback.
        """
        db_path = tmp_path / "failure_concrete_test.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            run_migrations(engine)
            rev_before = get_current_revision(engine)
            assert rev_before is not None

            # Pre-create the witness table outside the transaction the
            # patched call will run in.
            with engine.begin() as conn:
                conn.execute(text("CREATE TABLE rollback_witness (x INTEGER)"))

            with patch(
                "local_deep_research.database.alembic_runner.command.upgrade",
                side_effect=_insert_witness_then_fail,
            ):
                with pytest.raises(RuntimeError, match="simulated failure"):
                    run_migrations(engine, target=rev_before)

            assert get_current_revision(engine) == rev_before
            with engine.begin() as conn:
                rows = list(
                    conn.execute(text("SELECT x FROM rollback_witness"))
                )
            assert rows == []
        finally:
            engine.dispose()


class TestMigrationPathSecurity:
    """Tests for migration directory path security."""

    def test_migrations_dir_within_package(self):
        """Verify migrations directory is within expected package path."""
        migrations_dir = get_migrations_dir()
        real_path = migrations_dir.resolve()

        # Should be within the database package directory
        assert "local_deep_research" in str(real_path)
        assert "database" in str(real_path)
        assert "migrations" in str(real_path)

    def test_migrations_dir_is_not_symlink_by_default(self):
        """Verify the migrations directory is not a symlink by default."""
        migrations_dir = get_migrations_dir()

        # The directory itself should not be a symlink
        # (this documents expected behavior, actual symlink detection is in the function)
        assert migrations_dir.exists()

    @pytest.mark.skipif(
        os.name == "nt", reason="Symlink creation requires admin on Windows"
    )
    def test_symlink_attack_detection(self, tmp_path):
        """Verify symlink attacks are detected (mocked test)."""
        from pathlib import Path

        # Create a mock symlinked path scenario
        malicious_target = tmp_path / "malicious_migrations"
        malicious_target.mkdir()

        def mock_get_migrations_dir():
            """
            Simulates what get_migrations_dir() does but with a path
            that resolves outside the expected package boundary.
            """
            # The actual function should detect this
            real_path = malicious_target
            expected_parent = Path(__file__).parent.resolve()

            if not str(real_path).startswith(str(expected_parent)):
                raise ValueError(
                    "Invalid migrations path (possible symlink attack)"
                )

            return malicious_target

        # Verify that a simulated attack would be caught
        with pytest.raises(ValueError) as exc_info:
            mock_get_migrations_dir()

        assert "symlink attack" in str(exc_info.value)


class TestMigrationFilePermissions:
    """Tests for migration file permission validation."""

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_migration_files_not_world_writable(self):
        """Verify actual migration files are not world-writable."""
        migrations_dir = get_migrations_dir()
        versions_dir = migrations_dir / "versions"

        if not versions_dir.exists():
            pytest.skip("No versions directory exists yet")

        for migration_file in versions_dir.glob("*.py"):
            st = migration_file.stat()
            # Check world-writable bit is NOT set
            assert not (st.st_mode & 0o002), (
                f"Migration file {migration_file.name} is world-writable"
            )

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_world_writable_migration_rejected(self, tmp_path):
        """Verify world-writable migration files are rejected."""
        from local_deep_research.database.alembic_runner import (
            _validate_migrations_permissions,
        )

        # Create a mock migrations directory structure
        mock_migrations = tmp_path / "migrations"
        mock_versions = mock_migrations / "versions"
        mock_versions.mkdir(parents=True)

        # Create a world-writable migration file
        bad_migration = mock_versions / "001_bad_migration.py"
        bad_migration.write_text("# malicious content")
        os.chmod(bad_migration, 0o666)  # noqa: S103 — intentionally testing permission validation

        # Should raise ValueError
        with pytest.raises(ValueError) as exc_info:
            _validate_migrations_permissions(mock_migrations)

        assert "world-writable" in str(exc_info.value)
        assert "001_bad_migration.py" in str(exc_info.value)

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_secure_migration_permissions_accepted(self, tmp_path):
        """Verify migrations with secure permissions pass validation."""
        from local_deep_research.database.alembic_runner import (
            _validate_migrations_permissions,
        )

        # Create a mock migrations directory with secure files
        mock_migrations = tmp_path / "migrations"
        mock_versions = mock_migrations / "versions"
        mock_versions.mkdir(parents=True)

        # Create secure migration files
        good_migration = mock_versions / "001_good_migration.py"
        good_migration.write_text("# secure content")
        os.chmod(
            good_migration, 0o644
        )  # Owner rw, group r, other r (not writable)

        # Should not raise
        _validate_migrations_permissions(mock_migrations)

    def test_permission_check_skipped_on_windows(self, tmp_path, monkeypatch):
        """Verify permission checks are skipped on Windows."""
        from local_deep_research.database.alembic_runner import (
            _validate_migrations_permissions,
        )

        # Mock os.name to be 'nt' (Windows)
        monkeypatch.setattr(os, "name", "nt")

        # Create a mock migrations directory (permissions don't matter)
        mock_migrations = tmp_path / "migrations"
        mock_versions = mock_migrations / "versions"
        mock_versions.mkdir(parents=True)

        bad_migration = mock_versions / "001_migration.py"
        bad_migration.write_text("# content")

        # Should not raise even if file would be "insecure" on Unix
        _validate_migrations_permissions(mock_migrations)

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_world_writable_versions_dir_rejected(self, tmp_path):
        """Reject a world-writable versions/ directory even when files inside are secure.

        Regression: a Docker image landed with mode 0o777 on the directory
        itself (files were 0o644). Every per-user login then tripped this
        check, swallowed the resulting ValueError, and silently left the DB
        at its previous Alembic revision — surfacing later as
        ``no such table: papers`` on academic-source saves.
        """
        from local_deep_research.database.alembic_runner import (
            _validate_migrations_permissions,
        )

        mock_migrations = tmp_path / "migrations"
        mock_versions = mock_migrations / "versions"
        mock_versions.mkdir(parents=True)

        # File itself is fine — the only insecure thing is the directory.
        secure_migration = mock_versions / "001_migration.py"
        secure_migration.write_text("# content")
        os.chmod(secure_migration, 0o644)

        os.chmod(mock_versions, 0o777)  # noqa: S103 — intentionally testing rejection

        with pytest.raises(ValueError) as exc_info:
            _validate_migrations_permissions(mock_migrations)

        msg = str(exc_info.value)
        assert "world-writable" in msg
        assert "versions" in msg

    @pytest.mark.skipif(
        os.name == "nt", reason="File permissions work differently on Windows"
    )
    def test_shipped_versions_dir_not_world_writable(self):
        """The actual shipped versions/ directory must not be world-writable.

        Companion to ``test_migration_files_not_world_writable`` (which only
        covers the files). The runtime check rejects either; both need to
        be guarded against packaging regressions.
        """
        migrations_dir = get_migrations_dir()
        versions_dir = migrations_dir / "versions"
        if not versions_dir.exists():
            pytest.skip("No versions directory exists yet")

        st = versions_dir.stat()
        assert not (st.st_mode & 0o002), (
            f"versions/ directory is world-writable (mode={oct(st.st_mode)}). "
            "If this fires in CI after a packaging change, normalise the "
            "perms in Dockerfile/build, not by weakening this test."
        )


# =============================================================================
# Backward Compatibility Tests
# =============================================================================


class TestPreAlembicDatabaseUpgrade:
    """
    Simulate real-world upgrade scenarios where databases were created
    with the old Base.metadata.create_all() + _run_migrations() approach
    and must be seamlessly upgraded to Alembic-managed schema.
    """

    @pytest.fixture
    def pre_alembic_db_full(self, tmp_path):
        """
        Simulate a pre-Alembic database: all tables created via
        Base.metadata.create_all(), progress columns already added
        by the old _run_migrations() path. No alembic_version table.
        """
        db_path = tmp_path / "pre_alembic_full.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # This is exactly what the old initialize_database() did
        Base.metadata.create_all(engine, checkfirst=True)

        yield engine
        engine.dispose()

    @pytest.fixture
    def pre_alembic_db_old(self, tmp_path):
        """
        Simulate a very old pre-Alembic database: all tables created via
        Base.metadata.create_all(), but progress columns NOT yet added
        (simulates a database from before _run_migrations existed).
        """
        db_path = tmp_path / "pre_alembic_old.db"
        engine = create_engine(f"sqlite:///{db_path}")

        Base.metadata.create_all(engine, checkfirst=True)

        # Remove progress columns to simulate old schema
        with engine.begin() as conn:
            inspector = inspect(conn)
            if inspector.has_table("task_metadata"):
                cols = {
                    c["name"] for c in inspector.get_columns("task_metadata")
                }
                for col in [
                    "progress_current",
                    "progress_total",
                    "progress_message",
                    "metadata_json",
                ]:
                    if col in cols:
                        # SQLite doesn't support DROP COLUMN before 3.35,
                        # so recreate the table without those columns
                        pass
        # Simpler approach: create task_metadata manually without progress cols
        with engine.begin() as conn:
            conn.execute(text("DROP TABLE IF EXISTS task_metadata"))
            conn.execute(
                text("""
                CREATE TABLE task_metadata (
                    id INTEGER PRIMARY KEY,
                    task_id VARCHAR,
                    status VARCHAR,
                    task_type VARCHAR,
                    created_at DATETIME,
                    completed_at DATETIME,
                    duration_seconds FLOAT,
                    error_message VARCHAR
                )
            """)
            )

        yield engine
        engine.dispose()

    @pytest.fixture
    def pre_alembic_db_with_data(self, tmp_path):
        """
        Simulate a pre-Alembic database with realistic user data that
        must survive the migration.
        """
        db_path = tmp_path / "pre_alembic_data.db"
        engine = create_engine(f"sqlite:///{db_path}")

        Base.metadata.create_all(engine, checkfirst=True)

        # Insert realistic data using ORM (handles all NOT NULL defaults)
        from local_deep_research.database.models import Setting
        from local_deep_research.database.models.settings import SettingType

        Session = sessionmaker(bind=engine)
        with Session() as session:
            session.add(
                Setting(
                    key="llm.provider",
                    value="ollama",
                    type=SettingType.LLM,
                    name="LLM Provider",
                    category="llm",
                )
            )
            session.add(
                Setting(
                    key="search.engine",
                    value="duckduckgo",
                    type=SettingType.SEARCH,
                    name="Search Engine",
                    category="search",
                )
            )
            session.commit()

        # Use raw SQL for simpler tables without complex defaults
        with engine.begin() as conn:
            # Research history
            conn.execute(
                text("""
                INSERT INTO research_history
                    (id, query, mode, status, created_at, research_meta)
                VALUES
                    ('res-001', 'quantum computing', 'deep', 'completed',
                     datetime('now'), '{"iterations": 5}'),
                    ('res-002', 'machine learning', 'quick', 'completed',
                     datetime('now'), '{"iterations": 1}')
            """)
            )
            # Task metadata with progress columns (old _run_migrations added these)
            conn.execute(
                text("""
                INSERT INTO task_metadata
                    (task_id, status, task_type, progress_current, progress_total,
                     progress_message)
                VALUES
                    ('task-001', 'completed', 'research', 5, 5,
                     'Done'),
                    ('task-002', 'running', 'research', 2, 10,
                     'Iteration 2/10')
            """)
            )

        yield engine
        engine.dispose()

    def test_full_pre_alembic_db_upgrades_cleanly(self, pre_alembic_db_full):
        """A complete pre-Alembic database upgrades without errors."""
        engine = pre_alembic_db_full

        # Verify no alembic_version table yet
        inspector = inspect(engine)
        assert "alembic_version" not in inspector.get_table_names()

        # Run the new initialize_database (which calls run_migrations)
        initialize_database(engine)

        # Verify Alembic now tracks the database
        new_inspector = inspect(engine)
        assert "alembic_version" in new_inspector.get_table_names()

        # Verify at head revision
        assert get_current_revision(engine) == get_head_revision()
        assert not needs_migration(engine)

    def test_old_db_missing_progress_cols_gets_them_added(
        self, pre_alembic_db_old
    ):
        """
        A very old database missing progress columns gets them added
        via Alembic migration 0002.
        """
        engine = pre_alembic_db_old

        # Verify columns are missing
        inspector = inspect(engine)
        cols = {c["name"] for c in inspector.get_columns("task_metadata")}
        assert "progress_current" not in cols

        # Run migrations
        run_migrations(engine)  # raises on failure

        # Verify progress columns now exist
        new_inspector = inspect(engine)
        new_cols = {
            c["name"] for c in new_inspector.get_columns("task_metadata")
        }
        assert "progress_current" in new_cols
        assert "progress_total" in new_cols
        assert "progress_message" in new_cols
        assert "metadata_json" in new_cols

    def test_data_survives_full_upgrade_path(self, pre_alembic_db_with_data):
        """
        All user data (settings, research history, task metadata)
        survives the upgrade from pre-Alembic to Alembic-managed.
        """
        engine = pre_alembic_db_with_data

        # Run the full upgrade
        initialize_database(engine)

        # Verify settings preserved
        with engine.connect() as conn:
            settings = conn.execute(
                text("SELECT key, value FROM settings ORDER BY key")
            ).fetchall()
            settings_dict = {row[0]: row[1] for row in settings}
            assert settings_dict["llm.provider"] in ("ollama", '"ollama"')
            assert settings_dict["search.engine"] in (
                "duckduckgo",
                '"duckduckgo"',
            )

        # Verify research history preserved
        with engine.connect() as conn:
            research = conn.execute(
                text(
                    "SELECT id, query, status FROM research_history ORDER BY id"
                )
            ).fetchall()
            assert len(research) == 2
            assert research[0][1] == "quantum computing"
            assert research[1][1] == "machine learning"

        # Verify task metadata preserved (including progress columns)
        with engine.connect() as conn:
            tasks = conn.execute(
                text("""
                    SELECT task_id, status, progress_current, progress_total,
                           progress_message
                    FROM task_metadata ORDER BY task_id
                """)
            ).fetchall()
            assert len(tasks) == 2
            assert tasks[0][0] == "task-001"
            assert tasks[0][2] == 5  # progress_current
            assert tasks[0][3] == 5  # progress_total
            assert tasks[0][4] == "Done"
            assert tasks[1][0] == "task-002"
            assert tasks[1][2] == 2
            assert tasks[1][4] == "Iteration 2/10"

    def test_upgrade_then_reinitialize_is_idempotent(
        self, pre_alembic_db_with_data
    ):
        """
        After upgrading a pre-Alembic DB, calling initialize_database
        again should be a no-op (idempotent).
        """
        engine = pre_alembic_db_with_data

        # First upgrade
        initialize_database(engine)
        rev1 = get_current_revision(engine)

        inspector1 = inspect(engine)
        tables1 = set(inspector1.get_table_names())

        # Count data
        with engine.connect() as conn:
            settings_count = conn.execute(
                text("SELECT COUNT(*) FROM settings")
            ).fetchone()[0]
            research_count = conn.execute(
                text("SELECT COUNT(*) FROM research_history")
            ).fetchone()[0]

        # Second initialization (should be no-op)
        initialize_database(engine)
        rev2 = get_current_revision(engine)

        inspector2 = inspect(engine)
        tables2 = set(inspector2.get_table_names())

        # Verify nothing changed
        assert rev1 == rev2
        assert tables1 == tables2

        with engine.connect() as conn:
            assert (
                conn.execute(text("SELECT COUNT(*) FROM settings")).fetchone()[
                    0
                ]
                == settings_count
            )
            assert (
                conn.execute(
                    text("SELECT COUNT(*) FROM research_history")
                ).fetchone()[0]
                == research_count
            )

    def test_schema_matches_after_upgrade_vs_fresh(self, tmp_path):
        """
        The schema of an upgraded pre-Alembic database should match
        the schema of a freshly created Alembic database (minus the
        users table which pre-Alembic databases had).
        """
        # Create a fresh Alembic database
        fresh_path = tmp_path / "fresh.db"
        fresh_engine = create_engine(f"sqlite:///{fresh_path}")
        run_migrations(fresh_engine)

        # Create a pre-Alembic database and upgrade it
        legacy_path = tmp_path / "legacy.db"
        legacy_engine = create_engine(f"sqlite:///{legacy_path}")
        Base.metadata.create_all(legacy_engine, checkfirst=True)
        run_migrations(legacy_engine)

        try:
            fresh_inspector = inspect(fresh_engine)
            legacy_inspector = inspect(legacy_engine)

            fresh_tables = set(fresh_inspector.get_table_names())
            legacy_tables = set(legacy_inspector.get_table_names())

            # Legacy DB may have 'users' table (pre-Alembic created all tables)
            # Fresh Alembic DB skips 'users' (auth-only)
            legacy_tables.discard("users")

            assert fresh_tables == legacy_tables, (
                f"Table mismatch.\n"
                f"  Only in fresh: {fresh_tables - legacy_tables}\n"
                f"  Only in legacy: {legacy_tables - fresh_tables}"
            )

            # Compare columns for each shared table (excluding alembic_version)
            for table_name in fresh_tables - {"alembic_version"}:
                fresh_cols = {
                    c["name"] for c in fresh_inspector.get_columns(table_name)
                }
                legacy_cols = {
                    c["name"] for c in legacy_inspector.get_columns(table_name)
                }
                assert fresh_cols == legacy_cols, (
                    f"Column mismatch in table '{table_name}'.\n"
                    f"  Only in fresh: {fresh_cols - legacy_cols}\n"
                    f"  Only in legacy: {legacy_cols - fresh_cols}"
                )
        finally:
            fresh_engine.dispose()
            legacy_engine.dispose()

    def test_users_table_not_in_fresh_alembic_db(self, tmp_path):
        """
        Verify that a fresh Alembic-managed database does NOT contain
        the 'users' table (it's auth-only and created separately).
        """
        db_path = tmp_path / "fresh_no_users.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            run_migrations(engine)

            inspector = inspect(engine)
            tables = inspector.get_table_names()
            assert "users" not in tables
        finally:
            engine.dispose()

    def test_pre_alembic_db_with_users_table_keeps_it(self, tmp_path):
        """
        A pre-Alembic database that has a 'users' table (from the old
        create_all) should keep it after migration - Alembic doesn't
        drop tables it doesn't manage.
        """
        db_path = tmp_path / "legacy_with_users.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Old behavior: create_all creates everything including users
            Base.metadata.create_all(engine, checkfirst=True)

            inspector = inspect(engine)
            assert "users" in inspector.get_table_names()

            # Upgrade with Alembic
            run_migrations(engine)

            # users table should still be there (not dropped)
            new_inspector = inspect(engine)
            assert "users" in new_inspector.get_table_names()
        finally:
            engine.dispose()

    def test_multiple_sequential_upgrades(self, tmp_path):
        """
        Simulate the lifecycle: create DB → upgrade to 0001 → insert data
        → upgrade to 0002 → verify data + new columns.
        """
        db_path = tmp_path / "sequential.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Step 1: Migrate to 0001 only
            run_migrations(engine, target="0001")  # raises on failure
            assert get_current_revision(engine) == "0001"

            # Step 2: Insert data
            with engine.begin() as conn:
                conn.execute(
                    text("""
                    INSERT INTO task_metadata (task_id, status, task_type)
                    VALUES ('seq-001', 'running', 'research'),
                           ('seq-002', 'completed', 'research')
                """)
                )

            # Use ORM for settings (many NOT NULL columns with Python defaults)
            from local_deep_research.database.models import Setting
            from local_deep_research.database.models.settings import SettingType

            Session = sessionmaker(bind=engine)
            with Session() as session:
                session.add(
                    Setting(
                        key="test.key",
                        value="test_value",
                        type=SettingType.APP,
                        name="Test Key",
                        category="test",
                    )
                )
                session.commit()

            # Step 3: Upgrade to head (0005)
            run_migrations(engine, target="head")  # raises on failure
            assert get_current_revision(engine) == get_head_revision()

            # Step 4: Verify data survived
            with engine.connect() as conn:
                tasks = conn.execute(
                    text("SELECT task_id FROM task_metadata ORDER BY task_id")
                ).fetchall()
                assert len(tasks) == 2
                assert tasks[0][0] == "seq-001"

                settings = conn.execute(
                    text("SELECT value FROM settings WHERE key = 'test.key'")
                ).fetchone()
                assert settings[0] in ("test_value", '"test_value"')

            # Step 5: Verify new columns exist
            # Note: existing rows get NULL for new columns (SQLite batch alter
            # adds columns but doesn't backfill; server_default only applies
            # to future inserts)
            with engine.connect() as conn:
                row = conn.execute(
                    text("""
                        SELECT progress_current, progress_total
                        FROM task_metadata WHERE task_id = 'seq-001'
                    """)
                ).fetchone()
                assert row is not None  # Columns exist and are queryable

            # Verify new inserts can use the columns
            with engine.begin() as conn:
                conn.execute(
                    text("""
                    INSERT INTO task_metadata
                        (task_id, status, task_type, progress_current, progress_total)
                    VALUES ('seq-003', 'pending', 'research', 0, 5)
                """)
                )
            with engine.connect() as conn:
                row = conn.execute(
                    text("""
                        SELECT progress_current, progress_total
                        FROM task_metadata WHERE task_id = 'seq-003'
                    """)
                ).fetchone()
                assert row[0] == 0
                assert row[1] == 5

            # Step 6: Another run_migrations should be no-op
            run_migrations(engine)  # raises on failure
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestUpgradeFromBuggyV16xUserDatabase:
    """Regression for #3697.

    Pre-fix ``encrypted_db.create_user_database`` only emitted ``CreateTable``,
    never ``CreateIndex``. Combined with v1.6.0 enabling ``PRAGMA
    foreign_keys = ON``, every existing user DB has a mismatched FK
    (``download_attempts.url_hash`` → ``download_tracker.url_hash`` with no
    UNIQUE backing). On any DML touching those tables SQLite raises
    ``foreign key mismatch``, which would abort migration 0007 itself before
    it could repair the schema. Migration 0007 must therefore disable FK
    enforcement around its scrub + UNIQUE-index creation.
    """

    @pytest.fixture
    def buggy_v16x_engine(self, tmp_path):
        """Build a DB that mirrors the pre-fix state of an existing user:
        tables exist, but ``download_tracker.url_hash`` has no UNIQUE backing,
        the alembic_version is at 0006, and FK enforcement is on for every
        connection (mirroring ``apply_performance_pragmas``)."""
        import sqlite3
        from sqlalchemy import event

        db_path = tmp_path / "buggy_v16x.db"
        raw = sqlite3.connect(db_path)
        raw.execute("PRAGMA foreign_keys = OFF")
        raw.executescript(
            """
            CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY);
            INSERT INTO alembic_version VALUES ('0006');
            CREATE TABLE download_tracker (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                url_hash VARCHAR(64) NOT NULL,
                first_resource_id INTEGER,
                is_downloaded BOOLEAN,
                file_hash VARCHAR(64),
                file_path TEXT,
                file_name VARCHAR(255),
                file_size INTEGER,
                is_accessible BOOLEAN,
                first_seen TIMESTAMP,
                downloaded_at TIMESTAMP,
                last_checked TIMESTAMP,
                library_document_id INTEGER
            );
            CREATE TABLE download_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                attempt_number INTEGER,
                succeeded BOOLEAN,
                attempted_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE download_duplicates (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                resource_id INTEGER,
                research_id VARCHAR(36),
                added_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE research_history (
                id VARCHAR(36) PRIMARY KEY,
                query TEXT NOT NULL,
                mode TEXT NOT NULL,
                status TEXT NOT NULL,
                created_at TEXT NOT NULL,
                completed_at TEXT,
                duration_seconds INTEGER,
                report_path TEXT,
                report_content TEXT,
                research_meta TEXT,
                progress_log TEXT,
                progress INTEGER,
                title TEXT
            );
            """
        )
        # Seed: duplicate parent rows (HASH1) + valid child + orphan child.
        raw.execute(
            "INSERT INTO download_tracker (id, url, url_hash, first_resource_id) "
            "VALUES (1, 'a', 'HASH1', 1), (2, 'a-dup', 'HASH1', 2), (3, 'b', 'HASH2', 3)"
        )
        raw.execute(
            "INSERT INTO download_attempts (url_hash, attempt_number) "
            "VALUES ('HASH1', 1), ('HASH_ORPHAN', 1)"
        )
        raw.execute(
            "INSERT INTO download_duplicates (url_hash, resource_id, research_id) "
            "VALUES ('HASH_ORPHAN', 99, 'r1')"
        )
        raw.commit()
        raw.close()

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

        @event.listens_for(engine, "connect")
        def _enable_fk(dbapi_conn, _):
            dbapi_conn.execute("PRAGMA foreign_keys = ON")

        yield engine
        engine.dispose()

    def test_migration_0007_succeeds_against_pre_fix_user_db(
        self, buggy_v16x_engine
    ):
        """Without the FK toggle in 0007's upgrade(), the duplicate scrub
        raises ``foreign key mismatch`` and the migration aborts at 0006."""
        run_migrations(buggy_v16x_engine)
        # 0008 rides along on the same chain — assert head, not a literal.
        assert get_current_revision(buggy_v16x_engine) == get_head_revision()

        with buggy_v16x_engine.connect() as conn:
            assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1
            assert (
                conn.execute(text("PRAGMA foreign_key_check")).fetchall() == []
            )

            indexes = {
                row[1]
                for row in conn.execute(
                    text("PRAGMA index_list(download_tracker)")
                ).fetchall()
            }
            assert "uq_download_tracker_url_hash" in indexes

            tracker_ids = sorted(
                r[0]
                for r in conn.execute(
                    text("SELECT id FROM download_tracker")
                ).fetchall()
            )
            # Survivor of HASH1 group is min(id)=1; HASH2 unaffected.
            assert tracker_ids == [1, 3]

            attempt_hashes = {
                r[0]
                for r in conn.execute(
                    text("SELECT url_hash FROM download_attempts")
                ).fetchall()
            }
            assert attempt_hashes == {"HASH1"}  # orphan removed

            duplicate_count = conn.execute(
                text("SELECT COUNT(*) FROM download_duplicates")
            ).scalar()
            assert duplicate_count == 0  # orphan removed

    def test_fk_enforcement_active_after_migration(self, buggy_v16x_engine):
        """Once 0007 finishes, the repaired FK must actually reject inserts
        with a non-existent ``url_hash`` — proving FK was re-enabled and the
        UNIQUE backing is recognized."""
        from sqlalchemy.exc import IntegrityError

        run_migrations(buggy_v16x_engine)

        with buggy_v16x_engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO download_attempts (url_hash, attempt_number) "
                    "VALUES ('HASH1', 2)"
                )
            )

        with pytest.raises(IntegrityError):
            with buggy_v16x_engine.begin() as conn:
                conn.execute(
                    text(
                        "INSERT INTO download_attempts (url_hash, attempt_number) "
                        "VALUES ('NEW_ORPHAN', 1)"
                    )
                )


class TestUpgradeFromBuggyV16xUserDbProductionEngine:
    """Regression for #3990 — multi-migration upgrade fails with FK mismatch.

    The existing ``TestUpgradeFromBuggyV16xUserDatabase`` covers the same
    schema corruption but stamps at revision 0006, so migration 0007 is the
    first to run any DML in the upgrade transaction — which lets 0007's
    own ``PRAGMA foreign_keys = OFF`` actually take effect (no auto-begun
    transaction yet).

    Real production users (issue #3990) upgrade from 0001 to head. Migrations
    0002–0006 run DML before 0007, auto-beginning the sqlite3 driver
    transaction. ``PRAGMA foreign_keys`` is silently a no-op once a
    transaction is active (per sqlite.org/pragma.html#pragma_foreign_keys),
    so 0007's defensive PRAGMA never lands and the orphan-scrub DELETE
    fails with ``foreign key mismatch``.

    The fix (in ``alembic_runner.run_migrations``) issues PRAGMA OFF
    *before* opening the migration transaction. This test reproduces the
    production failure exactly: ``isolation_level=""`` (matching the
    sqlcipher3 engine in ``encrypted_db.py``) + FK ON at connect via the
    same event handler ``apply_performance_pragmas`` installs.
    """

    @pytest.fixture
    def buggy_v16x_production_engine(self, tmp_path):
        """Mirror the production engine: isolation_level="" + FK ON at
        connect, with the buggy v1.6.x schema stamped at revision 0005.

        Stamping at 0005 (not 0006) means migration 0006's data backfill
        (``UPDATE journals SET name_lower = ...``) runs DML before 0007,
        auto-beginning the driver transaction and freezing FK in the
        connect-time ON state for the rest of the upgrade.
        """
        import sqlite3

        from sqlalchemy import event

        db_path = tmp_path / "buggy_v16x_prod.db"
        # Seed the buggy v1.6.x schema with FK off (raw connection, no
        # FK target validation needed — the schema deliberately reflects
        # the pre-fix shape with no UNIQUE backing on download_tracker.url_hash).
        raw = sqlite3.connect(db_path)
        raw.execute("PRAGMA foreign_keys = OFF")
        raw.executescript(
            """
            CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY);
            INSERT INTO alembic_version VALUES ('0005');
            CREATE TABLE download_tracker (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                url_hash VARCHAR(64) NOT NULL,
                first_resource_id INTEGER,
                is_downloaded BOOLEAN,
                file_hash VARCHAR(64),
                file_path TEXT,
                file_name VARCHAR(255),
                file_size INTEGER,
                is_accessible BOOLEAN,
                first_seen TIMESTAMP,
                downloaded_at TIMESTAMP,
                last_checked TIMESTAMP,
                library_document_id INTEGER
            );
            CREATE TABLE download_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                attempt_number INTEGER,
                succeeded BOOLEAN,
                attempted_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE download_duplicates (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                resource_id INTEGER,
                research_id VARCHAR(36),
                added_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE journals (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name VARCHAR(500) NOT NULL,
                quality INTEGER,
                quality_analysis_time TIMESTAMP
            );
            INSERT INTO journals (name, quality) VALUES ('Nature', 100), ('NATURE', 95);
            """
        )
        raw.execute(
            "INSERT INTO download_tracker (id, url, url_hash, first_resource_id) "
            "VALUES (1, 'a', 'HASH1', 1), (2, 'a-dup', 'HASH1', 2), (3, 'b', 'HASH2', 3)"
        )
        raw.execute(
            "INSERT INTO download_attempts (url_hash, attempt_number) "
            "VALUES ('HASH1', 1), ('HASH_ORPHAN', 1)"
        )
        raw.execute(
            "INSERT INTO download_duplicates (url_hash, resource_id, research_id) "
            "VALUES ('HASH_ORPHAN', 99, 'r1')"
        )
        raw.commit()
        raw.close()

        # Production-shape engine: deferred isolation_level + FK ON at connect.
        def _create_conn():
            conn = sqlite3.connect(
                str(db_path),
                isolation_level="",
                check_same_thread=False,
            )
            conn.execute("PRAGMA foreign_keys = ON")
            return conn

        engine = create_engine("sqlite://", creator=_create_conn)

        @event.listens_for(engine, "connect")
        def _enable_fk(dbapi_conn, _):
            # Defensive: also fire on any pooled re-checkouts that bypass
            # the creator (matches encrypted_db's apply_performance_pragmas).
            dbapi_conn.execute("PRAGMA foreign_keys = ON")

        yield engine
        engine.dispose()

    def test_run_migrations_succeeds_through_full_chain(
        self, buggy_v16x_production_engine
    ):
        """Without the runner-level FK toggle, this fails with
        ``foreign key mismatch`` at 0007's orphan scrub."""
        run_migrations(buggy_v16x_production_engine)
        assert (
            get_current_revision(buggy_v16x_production_engine)
            == get_head_revision()
        )

        with buggy_v16x_production_engine.connect() as conn:
            # FK is back ON for the next checkout (engine was disposed).
            assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1
            # No FK violations remain in the repaired DB.
            assert (
                conn.execute(text("PRAGMA foreign_key_check")).fetchall() == []
            )
            # Orphan rows were scrubbed.
            attempt_hashes = {
                r[0]
                for r in conn.execute(
                    text("SELECT url_hash FROM download_attempts")
                ).fetchall()
            }
            assert attempt_hashes == {"HASH1"}
            duplicate_count = conn.execute(
                text("SELECT COUNT(*) FROM download_duplicates")
            ).scalar()
            assert duplicate_count == 0


class TestOrphanAlembicTempTableCleanup:
    """Regression for #3817 — ``table _alembic_tmp_journals already exists``.

    ``op.batch_alter_table`` rebuilds a table by creating
    ``_alembic_tmp_<table>``, copying data, dropping the original, and
    renaming. On a clean run alembic drops the temp table automatically.
    If a previous attempt failed in a way that bypassed transaction
    rollback (e.g., an older runner that auto-committed each migration),
    the temp table persists. The next ``batch_alter_table`` on the same
    parent fails with ``table _alembic_tmp_* already exists`` — even if
    the broader transaction would roll it back, alembic checks for
    pre-existing temp tables before creating its own.

    The fix drops orphan ``_alembic_tmp_*`` tables in
    ``alembic_runner.run_migrations`` before opening the migration
    transaction.
    """

    @pytest.fixture
    def db_with_orphan_temp_table(self, tmp_path):
        """Build a buggy v1.6.x DB at revision 0005 with an orphan
        ``_alembic_tmp_journals`` table left over from a prior crash."""
        import sqlite3

        db_path = tmp_path / "with_orphan_tmp.db"
        raw = sqlite3.connect(db_path)
        raw.execute("PRAGMA foreign_keys = OFF")
        raw.executescript(
            """
            CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY);
            INSERT INTO alembic_version VALUES ('0005');
            CREATE TABLE download_tracker (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT, url_hash VARCHAR(64) NOT NULL,
                first_resource_id INTEGER, is_downloaded BOOLEAN,
                file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255),
                file_size INTEGER, is_accessible BOOLEAN,
                first_seen TIMESTAMP, downloaded_at TIMESTAMP,
                last_checked TIMESTAMP, library_document_id INTEGER
            );
            CREATE TABLE download_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE download_duplicates (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE journals (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name VARCHAR(500) NOT NULL,
                quality INTEGER,
                quality_analysis_time TIMESTAMP
            );
            -- The smoking gun: a leftover batch_alter_table temp table
            -- from a prior crashed migration attempt. The schema can be
            -- arbitrary — alembic only checks the name.
            CREATE TABLE _alembic_tmp_journals (
                id INTEGER PRIMARY KEY,
                stale_marker TEXT
            );
            """
        )
        raw.commit()
        raw.close()

        def _create_conn():
            conn = sqlite3.connect(
                str(db_path), isolation_level="", check_same_thread=False
            )
            conn.execute("PRAGMA foreign_keys = ON")
            return conn

        engine = create_engine("sqlite://", creator=_create_conn)
        yield engine
        engine.dispose()

    def test_run_migrations_drops_orphan_temp_table(
        self, db_with_orphan_temp_table
    ):
        """Without the cleanup, migration 0006's batch_alter_table fails
        with ``table _alembic_tmp_journals already exists``."""
        # Sanity check: the orphan is present before the run.
        with db_with_orphan_temp_table.connect() as conn:
            tables = {
                r[0]
                for r in conn.execute(
                    text("SELECT name FROM sqlite_master WHERE type='table'")
                ).fetchall()
            }
        assert "_alembic_tmp_journals" in tables

        run_migrations(db_with_orphan_temp_table)

        assert (
            get_current_revision(db_with_orphan_temp_table)
            == get_head_revision()
        )

        # The orphan was dropped; alembic's own temp tables (if any from
        # this run) were also cleaned up by alembic itself.
        with db_with_orphan_temp_table.connect() as conn:
            tables_after = {
                r[0]
                for r in conn.execute(
                    text("SELECT name FROM sqlite_master WHERE type='table'")
                ).fetchall()
            }
        assert not any(t.startswith("_alembic_tmp_") for t in tables_after)

    @pytest.fixture
    def db_with_multiple_orphan_temp_tables(self, tmp_path):
        """Buggy v1.6.x DB at revision 0005 with THREE orphan
        ``_alembic_tmp_*`` tables — exercises the loop in
        ``_drop_orphan_alembic_temp_tables`` past index 0. A regression
        that replaces the iteration body with a single ``break`` (or
        otherwise short-circuits) would leave the trailing orphans
        behind and fail this test, while ``test_run_migrations_drops_orphan_temp_table``
        with its single orphan would still pass."""
        import sqlite3

        db_path = tmp_path / "with_multi_orphans.db"
        raw = sqlite3.connect(db_path)
        raw.execute("PRAGMA foreign_keys = OFF")
        raw.executescript(
            """
            CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY);
            INSERT INTO alembic_version VALUES ('0005');
            CREATE TABLE download_tracker (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT, url_hash VARCHAR(64) NOT NULL,
                first_resource_id INTEGER, is_downloaded BOOLEAN,
                file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255),
                file_size INTEGER, is_accessible BOOLEAN,
                first_seen TIMESTAMP, downloaded_at TIMESTAMP,
                last_checked TIMESTAMP, library_document_id INTEGER
            );
            CREATE TABLE download_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE download_duplicates (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url_hash VARCHAR(64) NOT NULL,
                resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP,
                FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash)
            );
            CREATE TABLE journals (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name VARCHAR(500) NOT NULL,
                quality INTEGER,
                quality_analysis_time TIMESTAMP
            );
            CREATE TABLE _alembic_tmp_journals (
                id INTEGER PRIMARY KEY, stale_marker TEXT
            );
            CREATE TABLE _alembic_tmp_research_history (
                id INTEGER PRIMARY KEY, stale_marker TEXT
            );
            CREATE TABLE _alembic_tmp_settings (
                id INTEGER PRIMARY KEY, stale_marker TEXT
            );
            """
        )
        raw.commit()
        raw.close()

        def _create_conn():
            conn = sqlite3.connect(
                str(db_path), isolation_level="", check_same_thread=False
            )
            conn.execute("PRAGMA foreign_keys = ON")
            return conn

        engine = create_engine("sqlite://", creator=_create_conn)
        yield engine
        engine.dispose()

    def test_run_migrations_drops_multiple_orphan_temp_tables(
        self, db_with_multiple_orphan_temp_tables
    ):
        """The cleanup must process every match — not just the first."""
        with db_with_multiple_orphan_temp_tables.connect() as conn:
            seeded = {
                r[0]
                for r in conn.execute(
                    text(
                        "SELECT name FROM sqlite_master "
                        "WHERE type='table' AND name LIKE '_alembic_tmp_%'"
                    )
                ).fetchall()
            }
        assert seeded == {
            "_alembic_tmp_journals",
            "_alembic_tmp_research_history",
            "_alembic_tmp_settings",
        }

        run_migrations(db_with_multiple_orphan_temp_tables)

        assert (
            get_current_revision(db_with_multiple_orphan_temp_tables)
            == get_head_revision()
        )
        with db_with_multiple_orphan_temp_tables.connect() as conn:
            remaining = [
                r[0]
                for r in conn.execute(
                    text(
                        "SELECT name FROM sqlite_master "
                        "WHERE type='table' AND name LIKE '_alembic_tmp_%'"
                    )
                ).fetchall()
            ]
        assert remaining == []

    def test_drop_orphan_temp_tables_no_op_when_none_present(
        self, tmp_path, loguru_caplog
    ):
        """Unit test on the cleanup helper itself: when the DB has no
        ``_alembic_tmp_*`` tables, the function must return without
        DDL and without emitting the ``Found N orphan alembic temp
        table(s)`` warning. This pins the early-return guard at the
        top of ``_drop_orphan_alembic_temp_tables`` so a future
        refactor that drops the guard (e.g. unconditional logging)
        would be caught immediately."""
        from local_deep_research.database.alembic_runner import (
            _drop_orphan_alembic_temp_tables,
        )

        db_path = tmp_path / "clean.db"
        engine = create_engine(f"sqlite:///{db_path}")
        try:
            with engine.connect() as conn:
                conn.exec_driver_sql(
                    "CREATE TABLE real_table (id INTEGER PRIMARY KEY)"
                )
                with loguru_caplog.at_level("WARNING"):
                    loguru_caplog.clear()
                    _drop_orphan_alembic_temp_tables(conn)
                assert (
                    "orphan alembic temp table"
                    not in loguru_caplog.text.lower()
                )
                # The real table is still there — we didn't touch anything.
                tables = {
                    r[0]
                    for r in conn.execute(
                        text(
                            "SELECT name FROM sqlite_master WHERE type='table'"
                        )
                    ).fetchall()
                }
                assert "real_table" in tables
        finally:
            engine.dispose()


class TestPreAlembicDatabaseHotfix:
    """Regression tests for bug #3747 — pre-Alembic DB login failure.

    Real users whose database was created before v1.4.0 (2026-03-25, when
    Alembic was introduced) have schema tables but no alembic_version row.
    Before this hotfix, run_migrations() called command.upgrade() from
    scratch, exposing legacy column shapes to migration 0007's index
    backfill and `download_tracker` scrub.

    The fixture uses raw-SQL legacy schema (NOT Base.metadata.create_all)
    so it reflects what a real pre-2026-03-21 user database actually looks
    like: `settings` omits the modern `category` column, `download_tracker`
    has the legacy shape without a UNIQUE constraint on `url_hash`, etc.
    """

    @pytest.fixture
    def pre_alembic_engine(self, tmp_path):
        engine = create_engine(f"sqlite:///{tmp_path}/pre_alembic.db")
        with engine.begin() as conn:
            # Legacy `settings` shape (no `category`, no `description`,
            # no `ui_element`, no enum extras — just the bare columns
            # that existed at project inception, 2025-06-29).
            conn.execute(
                text(
                    """
                    CREATE TABLE settings (
                        id INTEGER PRIMARY KEY,
                        key VARCHAR(255) NOT NULL UNIQUE,
                        value JSON,
                        type VARCHAR(50) NOT NULL,
                        name VARCHAR(255) NOT NULL,
                        visible BOOLEAN DEFAULT 1 NOT NULL,
                        editable BOOLEAN DEFAULT 1 NOT NULL,
                        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                        updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                    )
                    """
                )
            )
            conn.execute(
                text(
                    """
                    CREATE TABLE research_history (
                        id VARCHAR(36) PRIMARY KEY,
                        query TEXT NOT NULL,
                        mode VARCHAR(50),
                        status VARCHAR(50),
                        created_at DATETIME
                    )
                    """
                )
            )
            # Legacy `download_tracker` shape: url_hash without the
            # UNIQUE backing that migration 0007 expects to find. This
            # is the table that the unfixed migration path actually
            # trips over via 0007's index backfill / orphan scrub.
            conn.execute(
                text(
                    """
                    CREATE TABLE download_tracker (
                        id INTEGER PRIMARY KEY,
                        url TEXT NOT NULL,
                        url_hash VARCHAR(64) NOT NULL,
                        first_resource_id INTEGER,
                        is_downloaded BOOLEAN DEFAULT 0
                    )
                    """
                )
            )
        # Sanity: fixture really is pre-Alembic-shaped.
        inspector = inspect(engine)
        assert "alembic_version" not in inspector.get_table_names()
        assert "category" not in {
            c["name"] for c in inspector.get_columns("settings")
        }
        yield engine
        engine.dispose()

    def test_pre_alembic_db_reaches_head_and_stamp_branch_engaged(
        self, pre_alembic_engine, loguru_caplog
    ):
        """Core regression: pre-Alembic DB → run_migrations() → head, AND
        the BUG-3747 stamp branch is what got it there (not a coincidence)."""
        with loguru_caplog.at_level("WARNING"):
            run_migrations(pre_alembic_engine)
        assert get_current_revision(pre_alembic_engine) == get_head_revision()
        assert (
            "alembic_version" in inspect(pre_alembic_engine).get_table_names()
        )
        # Without the hotfix this log line never appears — its presence
        # proves we got to head via the stamp path, not by accident.
        assert "BUG-3747: pre-Alembic database detected" in loguru_caplog.text

    def test_pre_alembic_migration_is_idempotent(self, pre_alembic_engine):
        """Re-running run_migrations() after the hotfix is a no-op."""
        run_migrations(pre_alembic_engine)
        rev1 = get_current_revision(pre_alembic_engine)
        run_migrations(pre_alembic_engine)
        assert get_current_revision(pre_alembic_engine) == rev1

    def test_fresh_db_does_not_enter_stamp_branch(
        self, tmp_path, loguru_caplog
    ):
        """An empty DB must run upgrade from 0001, not the stamp branch."""
        engine = create_engine(f"sqlite:///{tmp_path}/fresh.db")
        try:
            with loguru_caplog.at_level("WARNING"):
                run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
            assert "BUG-3747" not in loguru_caplog.text
        finally:
            engine.dispose()

    def test_concurrent_stamp_is_neutralized(self, pre_alembic_engine):
        """stamp_database() must be safe to call twice (race-tolerant)."""
        stamp_database(pre_alembic_engine, "0001")
        # Second call simulates a racing concurrent caller — must NOT raise.
        stamp_database(pre_alembic_engine, "0001")
        assert get_current_revision(pre_alembic_engine) == "0001"

    def test_unrelated_operational_error_still_propagates(self, tmp_path):
        """The race-tolerance guard must NOT swallow non-race errors."""
        from alembic import command
        from sqlalchemy.exc import OperationalError

        engine = create_engine(f"sqlite:///{tmp_path}/will_error.db")
        original_stamp = command.stamp

        def _raise_unrelated(*_args, **_kwargs):
            raise OperationalError(
                "SELECT something", {}, Exception("disk I/O error")
            )

        try:
            command.stamp = _raise_unrelated
            with pytest.raises(OperationalError, match="disk I/O error"):
                stamp_database(engine, "0001")
        finally:
            command.stamp = original_stamp
            engine.dispose()

    def test_auth_db_shape_users_only_is_refused(self, tmp_path):
        """Engine with ONLY a `users` table must be refused (auth DB shape)."""
        engine = create_engine(f"sqlite:///{tmp_path}/auth.db")
        try:
            with engine.begin() as conn:
                conn.execute(
                    text(
                        "CREATE TABLE users "
                        "(id INTEGER PRIMARY KEY, username VARCHAR)"
                    )
                )
            with pytest.raises(RuntimeError, match="auth database"):
                run_migrations(engine)
        finally:
            engine.dispose()

    def test_auth_db_shape_users_plus_alembic_version_is_refused(
        self, tmp_path
    ):
        """Auth DB that's been (mis-)stamped must still be refused."""
        engine = create_engine(f"sqlite:///{tmp_path}/auth_stamped.db")
        try:
            with engine.begin() as conn:
                conn.execute(
                    text(
                        "CREATE TABLE users "
                        "(id INTEGER PRIMARY KEY, username VARCHAR)"
                    )
                )
                conn.execute(
                    text(
                        "CREATE TABLE alembic_version "
                        "(version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
                    )
                )
            with pytest.raises(RuntimeError, match="auth database"):
                run_migrations(engine)
        finally:
            engine.dispose()

    def test_pre_alembic_user_db_with_users_table_is_NOT_refused(
        self, tmp_path
    ):
        """Pre-Alembic user DBs contain `users` (created by the old
        `Base.metadata.create_all()` path before migration 0001 added
        the explicit skip). They must be allowed through, not refused
        as auth DBs.

        Uses `Base.metadata.create_all()` to simulate the legacy code
        path exactly — same approach as the existing pre-Alembic tests
        (see `test_pre_alembic_db_with_users_table_keeps_it`).
        """
        engine = create_engine(f"sqlite:///{tmp_path}/pre_alembic_user.db")
        try:
            Base.metadata.create_all(engine, checkfirst=True)
            # Sanity: the simulated pre-Alembic DB really does contain
            # `users` (the modern 0001 migration would have skipped it).
            assert "users" in inspect(engine).get_table_names()
            # Must NOT raise as an auth-DB false positive.
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()

    def test_only_alembic_version_table_is_treated_as_fresh(self, tmp_path):
        """A bare alembic_version (no schema, no users) is OK — runs upgrade."""
        engine = create_engine(f"sqlite:///{tmp_path}/bare.db")
        try:
            with engine.begin() as conn:
                conn.execute(
                    text(
                        "CREATE TABLE alembic_version "
                        "(version_num VARCHAR(32) NOT NULL "
                        "PRIMARY KEY)"
                    )
                )
            run_migrations(engine)
            assert get_current_revision(engine) == get_head_revision()
        finally:
            engine.dispose()


class TestMigrationAppliesSchemaChanges:
    """Tests proving the migration machinery actually changes schema and ORM works.

    These tests go beyond table/column name checks to verify:
    - Alembic Operations API works on our engine
    - Column properties (nullable, defaults) are correct after migration
    - ORM CRUD works after the production entry point
    - Downgrade/upgrade roundtrip preserves column properties
    """

    @pytest.fixture
    def migrated_to_head(self, tmp_path):
        """Engine with all migrations applied."""
        engine = create_engine(f"sqlite:///{tmp_path / 'head.db'}")
        run_migrations(engine, target="head")
        yield engine
        engine.dispose()

    def test_operations_api_can_modify_schema(self, migrated_to_head):
        """Alembic Operations API can create and drop a table on our engine."""
        from alembic.migration import MigrationContext
        from alembic.operations import Operations
        from sqlalchemy import Column, Integer, String

        with migrated_to_head.connect() as conn:
            ctx = MigrationContext.configure(conn)
            op = Operations(ctx)

            # Before: test table does NOT exist
            inspector = inspect(migrated_to_head)
            assert "_test_migration_probe" not in inspector.get_table_names()

            # Apply: create test table
            op.create_table(
                "_test_migration_probe",
                Column("id", Integer, primary_key=True),
                Column("name", String(50)),
            )

            # After: test table EXISTS with correct columns
            inspector = inspect(migrated_to_head)
            assert "_test_migration_probe" in inspector.get_table_names()
            cols = {
                c["name"]
                for c in inspector.get_columns("_test_migration_probe")
            }
            assert cols == {"id", "name"}

            # Cleanup: drop it
            op.drop_table("_test_migration_probe")
            inspector = inspect(migrated_to_head)
            assert "_test_migration_probe" not in inspector.get_table_names()

    def test_0011_adds_is_public_to_existing_collections(self, tmp_path):
        """Upgrade path: a DB created BEFORE 0011 (collections without
        is_public) gains the column on upgrade, and a pre-existing collection
        row is backfilled to private (False) via the server_default — existing
        users' collections don't become NULL/garbage or accidentally public.
        """
        from alembic import command

        engine = create_engine(f"sqlite:///{tmp_path / 'pre_0011.db'}")
        try:
            cfg = get_alembic_config(engine)

            # Simulate a pre-0011 DB: a collections table created by an older
            # version of the code (WITHOUT is_public), with the alembic
            # version stamped at 0010 so only 0011 runs. (The migration chain
            # can't reproduce this — 0001 builds from the *current* model,
            # which already has is_public — so we hand-create the old schema,
            # exactly as the 0002 test does.)
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                        CREATE TABLE collections (
                            id VARCHAR(36) PRIMARY KEY,
                            name VARCHAR(255) NOT NULL,
                            description TEXT,
                            collection_type VARCHAR(50),
                            is_default BOOLEAN,
                            created_at DATETIME,
                            updated_at DATETIME
                        )
                        """
                    )
                )
                conn.execute(
                    text(
                        "INSERT INTO collections (id, name) "
                        "VALUES ('legacy-c1', 'Legacy')"
                    )
                )

            with engine.begin() as conn:
                cfg.attributes["connection"] = conn
                command.stamp(cfg, "0010")

            cols = {
                c["name"] for c in inspect(engine).get_columns("collections")
            }
            assert "is_public" not in cols

            # Upgrade to head — 0011 adds is_public.
            with engine.begin() as conn:
                cfg.attributes["connection"] = conn
                command.upgrade(cfg, "head")

            col_map = {
                c["name"]: c for c in inspect(engine).get_columns("collections")
            }
            assert "is_public" in col_map
            assert col_map["is_public"]["nullable"] is True

            # The pre-existing row is backfilled to private (falsy).
            with engine.connect() as conn:
                val = conn.execute(
                    text(
                        "SELECT is_public FROM collections WHERE id='legacy-c1'"
                    )
                ).scalar()
                assert not val, (
                    f"existing row not private after upgrade: {val!r}"
                )
        finally:
            engine.dispose()

    def test_0002_adds_columns_with_properties_to_old_schema(self, tmp_path):
        """Migration 0002 adds progress columns with correct properties to a pre-Alembic schema.

        The 0001 migration creates all tables from Base.metadata (which includes
        progress columns). The 0002 migration is designed for pre-Alembic databases
        where task_metadata was created WITHOUT progress columns. This test
        simulates that scenario: stamp at 0001, manually create the old schema
        without progress columns, then run 0002 and verify column properties.
        """
        engine = create_engine(f"sqlite:///{tmp_path / 'old_schema_props.db'}")
        try:
            # Create task_metadata WITHOUT progress columns (simulating old schema)
            with engine.begin() as conn:
                conn.execute(
                    text(
                        """
                        CREATE TABLE task_metadata (
                            task_id VARCHAR PRIMARY KEY,
                            status VARCHAR NOT NULL,
                            task_type VARCHAR NOT NULL,
                            created_at DATETIME,
                            started_at DATETIME,
                            completed_at DATETIME,
                            error_message VARCHAR,
                            priority INTEGER DEFAULT 0,
                            retry_count INTEGER DEFAULT 0,
                            max_retries INTEGER DEFAULT 3
                        )
                        """
                    )
                )

            # Run the full migration — 0001 will create remaining tables,
            # and 0002 will add the missing progress columns
            run_migrations(engine)

            # AFTER: verify columns exist with correct properties
            inspector = inspect(engine)
            col_map = {
                c["name"]: c for c in inspector.get_columns("task_metadata")
            }

            assert "progress_current" in col_map
            assert "progress_total" in col_map
            assert "progress_message" in col_map
            assert "metadata_json" in col_map

            # Verify nullable constraints
            assert col_map["progress_message"]["nullable"] is True
            assert col_map["metadata_json"]["nullable"] is True

            # Verify defaults work via INSERT without specifying progress columns
            with engine.begin() as conn:
                conn.execute(
                    text(
                        "INSERT INTO task_metadata (task_id, status, task_type) "
                        "VALUES ('default-test', 'queued', 'research')"
                    )
                )
                row = conn.execute(
                    text(
                        "SELECT progress_current, progress_total FROM task_metadata "
                        "WHERE task_id='default-test'"
                    )
                ).fetchone()
                assert row[0] == 0  # default
                assert row[1] == 0  # default
        finally:
            engine.dispose()

    def test_orm_crud_after_initialize_database(self, tmp_path):
        """ORM CRUD works after initialize_database() — the production entry point."""
        from local_deep_research.database.models import TaskMetadata, Setting
        from local_deep_research.database.models.settings import SettingType

        engine = create_engine(f"sqlite:///{tmp_path / 'crud.db'}")
        try:
            Session = sessionmaker(bind=engine)

            # Production entry point
            with Session() as session:
                initialize_database(engine, session)

            # Verify migrations ran
            assert get_current_revision(engine) == get_head_revision()

            with Session() as session:
                # CREATE — TaskMetadata with 0002 progress columns
                task = TaskMetadata(
                    task_id="orm-test-001",
                    status="queued",
                    task_type="research",
                    progress_current=5,
                    progress_total=100,
                    progress_message="Searching...",
                    metadata_json={"source": "test"},
                )
                session.add(task)
                session.commit()

                # READ — verify all fields round-trip
                loaded = (
                    session.query(TaskMetadata)
                    .filter_by(task_id="orm-test-001")
                    .first()
                )
                assert loaded is not None
                assert loaded.progress_current == 5
                assert loaded.progress_total == 100
                assert loaded.progress_message == "Searching..."
                assert loaded.metadata_json == {"source": "test"}

                # UPDATE
                loaded.progress_current = 50
                loaded.progress_message = "Halfway"
                session.commit()
                reloaded = (
                    session.query(TaskMetadata)
                    .filter_by(task_id="orm-test-001")
                    .first()
                )
                assert reloaded.progress_current == 50
                assert reloaded.progress_message == "Halfway"

                # CREATE — Setting model (value is JSON type, key must be unique)
                setting = Setting(
                    key="test.migration.orm",
                    value="works",
                    type=SettingType.APP,
                    name="Test Setting",
                    category="test",
                )
                session.add(setting)
                session.commit()
                loaded_setting = (
                    session.query(Setting)
                    .filter_by(key="test.migration.orm")
                    .first()
                )
                assert loaded_setting.value == "works"

                # DELETE
                session.delete(reloaded)
                session.commit()
                assert (
                    session.query(TaskMetadata)
                    .filter_by(task_id="orm-test-001")
                    .first()
                    is None
                )
        finally:
            engine.dispose()

    def test_downgrade_upgrade_roundtrip_verifies_column_properties(
        self, tmp_path
    ):
        """Roundtrip downgrade/upgrade preserves column properties and ORM works.

        Starts at 0008 (last fully-reversible) rather than head; 0010
        is intentionally non-reversible (SQLite ALTER TABLE limitations).
        Re-upgrade to head still verifies forward path through 0010.
        """
        from alembic import command
        from local_deep_research.database.models import TaskMetadata

        migrated_to_head = create_engine(
            f"sqlite:///{tmp_path / 'roundtrip_head.db'}"
        )

        try:
            # Upgrade to 0008 (last reversible)
            config = get_alembic_config(migrated_to_head)
            with migrated_to_head.begin() as conn:
                config.attributes["connection"] = conn
                command.upgrade(config, "0008")

            # Insert data at 0008
            with migrated_to_head.begin() as conn:
                conn.execute(
                    text(
                        "INSERT INTO task_metadata (task_id, status, task_type, progress_current) "
                        "VALUES ('roundtrip', 'queued', 'research', 42)"
                    )
                )

            # DOWNGRADE to 0001
            with migrated_to_head.begin() as conn:
                config.attributes["connection"] = conn
                command.downgrade(config, "0001")

            # After downgrade: progress columns gone, core data kept
            inspector = inspect(migrated_to_head)
            cols = {c["name"] for c in inspector.get_columns("task_metadata")}
            assert "progress_current" not in cols
            assert "task_id" in cols

            with migrated_to_head.connect() as conn:
                row = conn.execute(
                    text(
                        "SELECT task_id FROM task_metadata WHERE task_id='roundtrip'"
                    )
                ).fetchone()
                assert row is not None

            # RE-UPGRADE to head (passes through 0010 in forward direction)
            run_migrations(migrated_to_head, target="head")

            # After re-upgrade: verify column PROPERTIES (not just names)
            inspector = inspect(migrated_to_head)
            col_map = {
                c["name"]: c for c in inspector.get_columns("task_metadata")
            }
            assert col_map["progress_message"]["nullable"] is True
            assert col_map["metadata_json"]["nullable"] is True

            # Verify defaults still work after roundtrip
            with migrated_to_head.begin() as conn:
                conn.execute(
                    text(
                        "INSERT INTO task_metadata (task_id, status, task_type) "
                        "VALUES ('after-roundtrip', 'queued', 'research')"
                    )
                )
                row = conn.execute(
                    text(
                        "SELECT progress_current, progress_total FROM task_metadata "
                        "WHERE task_id='after-roundtrip'"
                    )
                ).fetchone()
                assert row[0] == 0
                assert row[1] == 0

            # ORM CRUD works after roundtrip
            Session = sessionmaker(bind=migrated_to_head)
            with Session() as session:
                task = TaskMetadata(
                    task_id="post-roundtrip",
                    status="queued",
                    task_type="research",
                    progress_current=10,
                    progress_total=50,
                )
                session.add(task)
                session.commit()
                loaded = (
                    session.query(TaskMetadata)
                    .filter_by(task_id="post-roundtrip")
                    .first()
                )
                assert loaded.progress_current == 10
        finally:
            migrated_to_head.dispose()


# ---------------------------------------------------------------------------
# Helper for parametrized safety-guard tests
# ---------------------------------------------------------------------------


def _get_revision_chain():
    """Return [(revision_id, down_revision), ...] ordered base → head.

    Resolved at import time so ``@pytest.mark.parametrize`` can consume it
    during test collection.
    """
    from alembic.config import Config
    from alembic.script import ScriptDirectory

    from local_deep_research.database.alembic_runner import get_migrations_dir

    cfg = Config()
    cfg.set_main_option("script_location", str(get_migrations_dir()))
    script = ScriptDirectory.from_config(cfg)

    chain = []
    for rev in script.walk_revisions():
        down = rev.down_revision if rev.down_revision else "base"
        chain.append((rev.revision, down))
    chain.reverse()  # walk_revisions yields head-first
    return chain


_REVISION_CHAIN = _get_revision_chain()


def _get_migration_stems():
    """Return migration file stems for parametrized import tests.

    Resolved at import time so ``@pytest.mark.parametrize`` can consume it.
    """
    from local_deep_research.database.alembic_runner import get_migrations_dir

    versions_dir = get_migrations_dir() / "versions"
    return [p.stem for p in sorted(versions_dir.glob("[0-9]*.py"))]


_MIGRATION_STEMS = _get_migration_stems()


class TestNoCircularFkSawarning:
    """Regression: the documents ↔ research_resources circular FK used to
    emit a SAWarning from Base.metadata.sorted_tables on every cold start.

    Fixed by adding `use_alter=True` to ResearchResource.document_id so
    SQLAlchemy emits that one FK as a post-CREATE ALTER TABLE (breaking
    the dependency cycle for sorting purposes) while still creating the
    constraint at the database level.
    """

    def test_create_all_emits_no_circular_fk_warning(self, tmp_path):
        import warnings as _warnings

        engine = create_engine(f"sqlite:///{tmp_path}/test.db")
        try:
            with _warnings.catch_warnings(record=True) as caught:
                _warnings.simplefilter("always")
                Base.metadata.create_all(engine)
            offending = [
                w
                for w in caught
                if "unresolvable cycles" in str(w.message)
                and "documents" in str(w.message)
                and "research_resources" in str(w.message)
            ]
            assert not offending, (
                f"create_all() emitted the circular-FK SAWarning: "
                f"{[str(w.message) for w in offending]}"
            )
        finally:
            engine.dispose()


class TestMigrationSafetyGuards:
    """Guards that catch common migration pitfalls.

    These are *structural* checks — they don't test application logic, they
    prevent forgotten migrations, broken downgrades, branch conflicts, and
    orphaned models.
    """

    # -- fixtures -----------------------------------------------------------

    @pytest.fixture
    def fresh_engine(self, tmp_path):
        """Disposable in-memory style SQLite engine (file-backed for inspect)."""
        db_path = tmp_path / "guard_test.db"
        engine = create_engine(f"sqlite:///{db_path}")
        yield engine
        engine.dispose()

    @pytest.fixture
    def alembic_cfg(self, fresh_engine):
        """Alembic Config wired to *fresh_engine*."""
        return get_alembic_config(fresh_engine)

    # -- 1. schema drift ----------------------------------------------------

    def test_migrations_produce_schema_matching_models(
        self, fresh_engine, alembic_cfg
    ):
        """After upgrading to head the DB schema must match ORM metadata.

        Diffs that are *expected* and filtered out:
        * ``add_table('users')`` — the ``users`` table lives in Base.metadata
          but 0001 deliberately skips it (auth-only DB).
        * ``remove_index`` for the 9 performance indexes added by 0003 —
          those indexes are migration-only and intentionally absent from ORM
          model declarations.
        """
        from alembic import command
        from alembic.autogenerate import compare_metadata
        from alembic.runtime.migration import MigrationContext

        # Run all migrations on a clean DB
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.upgrade(alembic_cfg, "head")

        # Diff the resulting schema against ORM metadata
        with fresh_engine.connect() as conn:
            mc = MigrationContext.configure(conn)
            raw_diffs = compare_metadata(mc, Base.metadata)

        # Filter expected diffs
        unexpected = []
        for diff in raw_diffs:
            op_type = diff[0]

            # users table is intentionally excluded from migrations
            if op_type == "add_table" and diff[1].name == "users":
                continue
            # Index on users table (comes with the table definition)
            if op_type == "add_index" and diff[1].table.name == "users":
                continue

            # 0003 migration-only indexes show as "remove_index"
            if op_type == "remove_index":
                continue

            unexpected.append(diff)

        assert unexpected == [], (
            f"Schema drift detected — {len(unexpected)} diff(s) between "
            f"migrations and models:\n" + "\n".join(str(d) for d in unexpected)
        )

    # -- 2. single head -----------------------------------------------------

    def test_single_head_revision(self):
        """There must be exactly one head revision (no branch conflicts)."""
        from alembic.config import Config as AlembicConfig
        from alembic.script import ScriptDirectory

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        cfg = AlembicConfig()
        cfg.set_main_option("script_location", str(get_migrations_dir()))
        script = ScriptDirectory.from_config(cfg)
        heads = script.get_heads()

        assert len(heads) == 1, (
            f"Expected exactly 1 head revision, found {len(heads)}: {heads}. "
            "This means two migrations share the same parent — "
            "resolve with `alembic merge`."
        )

    # -- 3. stairway (up-down-up per revision) ------------------------------

    @pytest.mark.parametrize(
        "revision,down_rev",
        _REVISION_CHAIN,
        ids=[r[0] for r in _REVISION_CHAIN],
    )
    def test_stairway_up_down_up_per_revision(
        self, fresh_engine, alembic_cfg, revision, down_rev
    ):
        """Each revision must survive: parent → up → down → up."""
        from alembic import command

        if revision in self.NON_REVERSIBLE_REVISIONS:
            pytest.skip(
                f"Revision {revision} is intentionally non-reversible "
                "(SQLite ALTER TABLE limitations); see NON_REVERSIBLE_REVISIONS."
            )

        target_down = down_rev if down_rev != "base" else "base"
        expected_after_down = None if down_rev == "base" else down_rev

        # Upgrade to parent first (unless this is the base migration)
        if down_rev != "base":
            with fresh_engine.begin() as conn:
                alembic_cfg.attributes["connection"] = conn
                command.upgrade(alembic_cfg, down_rev)

        # Up
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.upgrade(alembic_cfg, revision)
        assert get_current_revision(fresh_engine) == revision

        # Down
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.downgrade(alembic_cfg, target_down)
        assert get_current_revision(fresh_engine) == expected_after_down

        # Up again
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.upgrade(alembic_cfg, revision)
        assert get_current_revision(fresh_engine) == revision

    # -- 4. substantive downgrades ------------------------------------------

    # Migrations whose downgrade is intentionally a no-op (e.g. one-time
    # data migrations that delete stale keys with no consumers).
    INTENTIONAL_NOOP_DOWNGRADES = {
        "0004_migrate_legacy_app_settings.py",
        # 0013: restoring 'auto'/'parallel' references would recreate
        # broken state — those engines no longer exist in the codebase.
        "0013_remove_meta_search_engines.py",
        # 0016: the dropped cache tables were orphaned dead code holding no
        # data, and their models are removed — nothing to recreate.
        "0016_drop_orphaned_cache_tables.py",
        # 0018: restoring 'mcp'/'agentic' strategy references would recreate
        # broken state — that strategy no longer exists in the codebase.
        "0018_remove_mcp_strategy.py",
        # 0019: the 'both' -> 'adaptive' coercion is lossy — after upgrade a
        # row is indistinguishable from an originally-'adaptive' one, so a
        # downgrade cannot know which rows to restore. Rewriting 'adaptive'
        # back to 'both' would clobber the legitimate default. No-op by design.
        "0019_retire_both_egress_scope.py",
    }

    # Migrations whose downgrade is intentionally NotImplementedError
    # (e.g. SQLite ALTER TABLE limitations against legacy unnamed
    # constraints + FK-target columns). These revisions are exempt from
    # parametrized stairway/residual tests and tested separately.
    NON_REVERSIBLE_REVISIONS = {
        "0010",  # 0010_add_chat_tables.py — chat schema, dev-stage rollback path: recreate DB
    }

    def test_all_downgrades_are_substantive(self):
        """Every migration's ``downgrade()`` must contain real operations.

        A ``pass``-only or empty downgrade silently blocks rollback.
        Migrations listed in INTENTIONAL_NOOP_DOWNGRADES are exempt.
        """
        import ast

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        versions_dir = get_migrations_dir() / "versions"

        for py_file in sorted(versions_dir.glob("[0-9]*.py")):
            if py_file.name in self.INTENTIONAL_NOOP_DOWNGRADES:
                continue

            tree = ast.parse(py_file.read_text())

            for node in ast.walk(tree):
                if not isinstance(node, ast.FunctionDef):
                    continue
                if node.name != "downgrade":
                    continue

                # Strip docstrings — a single Expr(Constant(str)) at position 0
                body = list(node.body)
                if (
                    body
                    and isinstance(body[0], ast.Expr)
                    and isinstance(body[0].value, ast.Constant)
                    and isinstance(body[0].value.value, str)
                ):
                    body = body[1:]

                # Must have at least one non-pass statement
                non_pass = [s for s in body if not isinstance(s, ast.Pass)]
                assert non_pass, (
                    f"{py_file.name}: downgrade() is empty or pass-only — "
                    "this migration cannot be rolled back."
                )

    # -- 5. all models registered on metadata -------------------------------

    def test_all_models_registered_on_metadata(self):
        """Every model submodule must be imported so Alembic can see its tables."""
        import importlib
        import pkgutil

        import local_deep_research.database.models as models_pkg

        # Force-import every submodule under the models package
        for _importer, modname, _ispkg in pkgutil.walk_packages(
            models_pkg.__path__,
            prefix=models_pkg.__name__ + ".",
        ):
            importlib.import_module(modname)

        # Critical tables that must exist in metadata
        critical_tables = {
            "settings",
            "research_tasks",
            "research_history",
            "task_metadata",
            "queue_status",
            "benchmark_runs",
            "benchmark_results",
            "benchmark_configs",
            "benchmark_progress",
            "token_usage",
            "search_calls",
            "reports",
            "report_sections",
            "app_logs",
            "provider_models",
            "documents",
            "collections",
        }

        registered = set(Base.metadata.tables.keys())
        missing = critical_tables - registered

        assert not missing, (
            f"Tables missing from Base.metadata (model not imported?): "
            f"{sorted(missing)}"
        )

    # -- 6. downgrade leaves no residual tables ----------------------------

    @pytest.mark.parametrize(
        "revision,down_rev",
        _REVISION_CHAIN,
        ids=[r[0] for r in _REVISION_CHAIN],
    )
    def test_downgrade_leaves_no_residual_tables(
        self, fresh_engine, alembic_cfg, revision, down_rev
    ):
        """Downgrading a revision must not leave behind tables it created."""
        from alembic import command

        if revision in self.NON_REVERSIBLE_REVISIONS:
            pytest.skip(
                f"Revision {revision} is intentionally non-reversible "
                "(SQLite ALTER TABLE limitations); see NON_REVERSIBLE_REVISIONS."
            )

        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn

            # Upgrade to parent first
            if down_rev != "base":
                command.upgrade(alembic_cfg, down_rev)

        tables_before = set(inspect(fresh_engine).get_table_names())

        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            # Upgrade to this revision
            command.upgrade(alembic_cfg, revision)

        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            # Downgrade back
            target_down = down_rev if down_rev != "base" else "base"
            command.downgrade(alembic_cfg, target_down)

        tables_after = set(inspect(fresh_engine).get_table_names())

        # alembic_version may appear/disappear — that's fine
        residual = (tables_after - tables_before) - {"alembic_version"}

        assert not residual, (
            f"Revision {revision}: downgrade left residual tables: "
            f"{sorted(residual)}"
        )

    # -- 7. deterministic schema --------------------------------------------

    def test_deterministic_schema(self, tmp_path):
        """Two independent fresh databases must produce identical schemas.

        Catches migrations that use timestamps, random values, or
        environment-dependent conditional logic.
        """
        from alembic import command

        schemas = []
        for i in range(2):
            db_path = tmp_path / f"deterministic_{i}.db"
            engine = create_engine(f"sqlite:///{db_path}")
            try:
                cfg = get_alembic_config(engine)
                with engine.begin() as conn:
                    cfg.attributes["connection"] = conn
                    command.upgrade(cfg, "head")

                insp = inspect(engine)
                schema = {}
                for table_name in sorted(insp.get_table_names()):
                    cols = [
                        (c["name"], str(c["type"]))
                        for c in insp.get_columns(table_name)
                    ]
                    idxs = sorted(
                        [
                            (idx["name"], tuple(idx["column_names"]))
                            for idx in insp.get_indexes(table_name)
                        ]
                    )
                    schema[table_name] = {"columns": cols, "indexes": idxs}
                schemas.append(schema)
            finally:
                engine.dispose()

        assert schemas[0] == schemas[1], (
            "Two fresh databases produced different schemas — "
            "a migration is non-deterministic."
        )

    # -- 8. each migration file is importable -------------------------------

    @pytest.mark.parametrize("stem", _MIGRATION_STEMS)
    def test_each_migration_revision_is_importable(self, stem):
        """Every migration file must import without SyntaxError or ImportError."""
        import importlib

        mod = importlib.import_module(
            f"local_deep_research.database.migrations.versions.{stem}"
        )
        # Sanity: the module must expose upgrade/downgrade callables
        assert callable(getattr(mod, "upgrade", None)), (
            f"{stem}: missing upgrade()"
        )
        assert callable(getattr(mod, "downgrade", None)), (
            f"{stem}: missing downgrade()"
        )

    # -- 9. downgrade data loss is explicit (0002) --------------------------

    def test_downgrade_data_loss_is_explicit_0002(
        self, fresh_engine, alembic_cfg
    ):
        """0002's downgrade drops progress columns; re-upgrade gets defaults, not old data."""
        from alembic import command

        # Migrate to 0002
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.upgrade(alembic_cfg, "0002")

        # Insert data into the progress columns
        with fresh_engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO task_metadata "
                    "(task_id, task_type, status, progress_current, progress_total, progress_message) "
                    "VALUES (:tid, :tt, :st, :pc, :pt, :pm)"
                ),
                {
                    "tid": "test-task-42",
                    "tt": "research",
                    "st": "running",
                    "pc": 7,
                    "pt": 10,
                    "pm": "Step 7 of 10",
                },
            )

        # Downgrade to 0001
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.downgrade(alembic_cfg, "0001")

        # Verify progress columns are gone
        cols = {
            c["name"]
            for c in inspect(fresh_engine).get_columns("task_metadata")
        }
        assert "progress_current" not in cols
        assert "progress_total" not in cols
        assert "progress_message" not in cols

        # Re-upgrade to 0002
        with fresh_engine.begin() as conn:
            alembic_cfg.attributes["connection"] = conn
            command.upgrade(alembic_cfg, "0002")

        # Columns are back with defaults, old data is gone
        with fresh_engine.connect() as conn:
            row = conn.execute(
                text(
                    "SELECT progress_current, progress_total, progress_message "
                    "FROM task_metadata WHERE task_id = :tid"
                ),
                {"tid": "test-task-42"},
            ).fetchone()

        assert row is not None, (
            "Core row should survive the downgrade/upgrade cycle"
        )
        # Defaults: 0, 0, NULL — the old values (7, 10, "Step 7 of 10") are lost
        assert row[0] == 0, (
            f"progress_current should be default 0, got {row[0]}"
        )
        assert row[1] == 0, f"progress_total should be default 0, got {row[1]}"
        assert row[2] is None, (
            f"progress_message should be default NULL, got {row[2]}"
        )

    # -- 10. env.py offline mode raises -------------------------------------

    def test_env_offline_mode_raises(self):
        """env.py's run_migrations_offline() must raise NotImplementedError.

        Tested via AST to avoid import side-effects (env.py runs
        run_migrations_online() at module level).
        """
        import ast

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        env_path = get_migrations_dir() / "env.py"
        tree = ast.parse(env_path.read_text(), filename=str(env_path))

        func = None
        for node in ast.walk(tree):
            if (
                isinstance(node, ast.FunctionDef)
                and node.name == "run_migrations_offline"
            ):
                func = node
                break

        assert func is not None, "run_migrations_offline not found in env.py"

        # Strip docstring
        body = list(func.body)
        if (
            body
            and isinstance(body[0], ast.Expr)
            and isinstance(body[0].value, ast.Constant)
            and isinstance(body[0].value.value, str)
        ):
            body = body[1:]

        assert len(body) == 1, (
            f"run_migrations_offline should contain only a Raise, "
            f"got {len(body)} statement(s)"
        )
        stmt = body[0]
        assert isinstance(stmt, ast.Raise), (
            f"Expected Raise statement, got {type(stmt).__name__}"
        )
        # Verify it raises NotImplementedError specifically
        assert isinstance(stmt.exc, ast.Call), (
            "Expected Raise with a Call (e.g. NotImplementedError(...))"
        )
        assert isinstance(stmt.exc.func, ast.Name), (
            "Expected exception to be a named type"
        )
        assert stmt.exc.func.id == "NotImplementedError", (
            f"Expected NotImplementedError, got {stmt.exc.func.id}"
        )

    # -- 10b. env.py online mode requires connection -------------------------

    def test_env_online_mode_requires_connection(self):
        """env.py's run_migrations_online() must raise RuntimeError when
        config.attributes['connection'] is None.

        Tested via AST to avoid import side-effects (env.py runs
        run_migrations_online() at module level).
        """
        import ast

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        env_path = get_migrations_dir() / "env.py"
        tree = ast.parse(env_path.read_text(), filename=str(env_path))

        func = None
        for node in ast.walk(tree):
            if (
                isinstance(node, ast.FunctionDef)
                and node.name == "run_migrations_online"
            ):
                func = node
                break

        assert func is not None, "run_migrations_online not found in env.py"

        # Find the RuntimeError raise in the function body
        has_runtime_error = False
        for node in ast.walk(func):
            if isinstance(node, ast.Raise) and isinstance(node.exc, ast.Call):
                if (
                    isinstance(node.exc.func, ast.Name)
                    and node.exc.func.id == "RuntimeError"
                ):
                    has_runtime_error = True
                    break

        assert has_runtime_error, (
            "run_migrations_online must raise RuntimeError when no "
            "connection is provided"
        )

    # -- 10c. migration 0004 guards on missing settings table ---------------

    def test_migration_0004_skips_without_settings_table(self, tmp_path):
        """Migration 0004 should succeed silently on a database without
        a settings table (e.g., if only partial tables exist).

        Target is 0008 (last revision that doesn't require a fully-formed
        research_history schema). 0010 ADDs a column to research_history
        which the partial-DB shape doesn't have; tested separately.
        """
        from local_deep_research.database.alembic_runner import (
            get_alembic_config,
        )

        # Create engine with only alembic_version stamped at 0003
        db_path = tmp_path / "no_settings.db"
        engine = create_engine(f"sqlite:///{db_path}")

        # Create alembic_version and stamp at 0003
        with engine.begin() as conn:
            conn.execute(
                text(
                    "CREATE TABLE alembic_version "
                    "(version_num VARCHAR(32) NOT NULL)"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO alembic_version (version_num) VALUES ('0003')"
                )
            )

        # Running migrations should not fail — 0004 should skip gracefully
        config = get_alembic_config(engine)
        from alembic import command

        with engine.begin() as conn:
            config.attributes["connection"] = conn
            command.upgrade(config, "0008")

        # Verify we're at 0008 despite no settings table
        with engine.connect() as conn:
            row = conn.execute(
                text("SELECT version_num FROM alembic_version")
            ).fetchone()
            assert row[0] == "0008"

        engine.dispose()

    # -- 11. revision IDs match filenames -----------------------------------

    def test_migration_revision_ids_match_filenames(self):
        """Each migration file's ``revision`` variable must match its filename prefix.

        Catches copy-paste errors where a migration file is duplicated
        but the ``revision = "..."`` inside is not updated.
        """
        import ast

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        versions_dir = get_migrations_dir() / "versions"

        for py_file in sorted(versions_dir.glob("[0-9]*.py")):
            filename_prefix = py_file.stem.split("_")[0]  # e.g. "0003"
            tree = ast.parse(py_file.read_text(), filename=str(py_file))

            revision_value = None
            for node in ast.walk(tree):
                if isinstance(node, ast.Assign):
                    for target in node.targets:
                        if (
                            isinstance(target, ast.Name)
                            and target.id == "revision"
                        ):
                            if isinstance(node.value, ast.Constant):
                                revision_value = node.value.value
                            break

            assert revision_value is not None, (
                f"{py_file.name}: no 'revision = \"...\"' assignment found"
            )
            assert revision_value == filename_prefix, (
                f"{py_file.name}: revision='{revision_value}' does not match "
                f"filename prefix '{filename_prefix}'"
            )