diff --git a/.github/workflows/quality-and-security.yml b/.github/workflows/quality-and-security.yml
index a242c92..558581f 100644
--- a/.github/workflows/quality-and-security.yml
+++ b/.github/workflows/quality-and-security.yml
@@ -25,6 +25,24 @@ jobs:
         # tests/test_supply_chain_policy.py keeps these three in agreement.
         python-version: ["3.12", "3.13", "3.14"]
 
+    # Each matrix worker owns an isolated server; fixtures create/drop only
+    # UUID databases, never the configured maintenance/application database.
+    services:
+      postgres:
+        image: postgres:18-bookworm
+        env:
+          POSTGRES_USER: scanner_test
+          POSTGRES_PASSWORD: scanner_ci_only
+          POSTGRES_DB: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U scanner_test -d postgres"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      SCANNER_TEST_POSTGRES_ADMIN_URL: postgresql+psycopg://scanner_test:scanner_ci_only@127.0.0.1:5432/postgres
+      SCANNER_REQUIRE_POSTGRES_TESTS: "1"
+
     steps:
       - name: Check out repository
         uses: actions/checkout@v7
@@ -47,6 +65,11 @@ jobs:
         run: python -m pre_commit validate-config .pre-commit-config.yaml
 
       - name: Run tests with coverage gate
+        env:
+          # runner.temp is available at step scope, after runner allocation.
+          # The private file lets Alembic and application engines share schema
+          # without borrowing configured application data or an in-memory DB.
+          DATABASE_URL: sqlite:///${{ runner.temp }}/scanner-default.db
         run: |
           # The floor sits ~two points under the measured 89% (QUAL-005) so
           # routine PRs don't fight the gate, while a large untested addition
@@ -114,6 +137,16 @@ jobs:
       - name: Smoke-test Compose stack
         run: docker compose up --build --wait --wait-timeout 180
 
+      - name: Verify application storage in final image
+        env:
+          # Template-only CI credentials; the helper creates its own databases.
+          SCANNER_TEST_POSTGRES_ADMIN_URL: postgresql+psycopg://scanner:scanner_dev_password_change_me@postgres:5432/postgres
+        run: >-
+          docker compose run --rm -T --no-deps
+          -e SCANNER_TEST_POSTGRES_ADMIN_URL -e DATABASE_URL=sqlite://
+          -e ALERT_ENABLED=false
+          scanner-ui python -m backend.storage.verification_smoke
+
       - name: Tear down Compose stack
         if: always()
         run: docker compose down --volumes --remove-orphans
diff --git a/backend/storage/schema_verification.py b/backend/storage/schema_verification.py
new file mode 100644
index 0000000..f42f6d1
--- /dev/null
+++ b/backend/storage/schema_verification.py
@@ -0,0 +1,163 @@
+"""Read-only, dialect-local schema fingerprints for migration verification.
+
+Beginner note:
+    Compare migrated and ORM-created databases on the SAME database dialect.
+    PostgreSQL rewrites expressions when storing them; reflecting both sides
+    lets its own parser normalize them without deleting meaningful SQL text.
+    This helper never repairs a schema or disposes the caller's engine.
+"""
+
+from __future__ import annotations
+
+import warnings
+from collections.abc import Mapping
+from typing import Any, cast
+
+from sqlalchemy import Boolean, inspect
+from sqlalchemy.engine import Engine
+from sqlalchemy.exc import SAWarning
+
+
+def _freeze(value: Any) -> Any:
+    """Make reflected options comparable without losing expressions or order.
+
+    Args:
+        value: SQLAlchemy reflection data, including dialect-specific clauses.
+
+    Returns:
+        An immutable value retaining mapping keys, sequence order and SQL text.
+
+    Beginner note:
+        SQL literals are case and whitespace sensitive. Do not lowercase text,
+        remove spaces or strip casts to make an unexplained discrepancy pass.
+    """
+    if isinstance(value, Mapping):
+        return tuple(sorted((key, _freeze(item)) for key, item in value.items()))
+    if isinstance(value, (list, tuple)):
+        return tuple(_freeze(item) for item in value)
+    if value is None or isinstance(value, (str, int, bool)):
+        return value
+    return str(value)
+
+
+def _sqlite_index_definitions(engine: Engine, table: str) -> frozenset[tuple[object, ...]]:
+    """Retain index key semantics that generic SQLite reflection leaves out.
+
+    Args:
+        engine: Caller-owned SQLite verification engine.
+        table: Reflected table name, quoted as an identifier before each PRAGMA.
+
+    Returns:
+        Index origins, uniqueness, partial-index flags, and ordered key/auxiliary
+        columns with their sort direction and collation. Explicit index names
+        remain significant; generated constraint-index names are excluded.
+
+    Raises:
+        sqlalchemy.exc.SQLAlchemyError: SQLite cannot return the index metadata.
+
+    Beginner note:
+        BINARY and NOCASE unique indexes accept different values even when their
+        names and columns match. ``index_xinfo`` exposes those comparison rules
+        and DESC order, which SQLAlchemy's generic reflection silently omits.
+        ``index_list`` also includes indexes backing UNIQUE and PRIMARY KEY
+        constraints. Their generated suffixes depend on declaration order, so
+        compare their actual definitions instead of incidental autoindex names.
+        See https://www.sqlite.org/pragma.html#pragma_index_xinfo.
+    """
+    quote = engine.dialect.identifier_preparer.quote_identifier
+    definitions: set[tuple[object, ...]] = set()
+    with engine.connect() as connection:
+        indexes = connection.exec_driver_sql(f"PRAGMA index_list({quote(table)})").all()
+        for index in indexes:
+            columns = connection.exec_driver_sql(f"PRAGMA index_xinfo({quote(index[1])})").all()
+            # Preserve index order and key/auxiliary status; table-column numeric
+            # offsets are incidental because the logical column name is retained.
+            keys = tuple((column[0], *column[2:]) for column in columns)
+            name = index[1] if index[3] == "c" else None
+            definitions.add((name, index[2], index[3], index[4], keys))
+    return frozenset(definitions)
+
+
+def reflect_schema(engine: Engine) -> dict[str, dict[str, object]]:
+    """Fingerprint database-enforced contracts, excluding Alembic bookkeeping.
+
+    Args:
+        engine: Caller-owned engine pointing at a verification database.
+
+    Returns:
+        Tables with columns, defaults, primary keys, checks, unique constraints,
+        complete index definitions and foreign-key targets/actions/options.
+
+    Raises:
+        sqlalchemy.exc.SQLAlchemyError: The caller's database cannot be queried
+            or its schema cannot be reflected.
+        sqlalchemy.exc.SAWarning: Index reflection is incomplete, for example
+            when the dialect cannot describe an expression index. Such a schema
+            must not be reported as matching from partial evidence.
+
+    Beginner note:
+        A matching index name says nothing about its columns or WHERE clause.
+        Index/check/unique names are retained. FK and PK names are excluded:
+        migrations name some SQLite batch-rebuilt FKs that the ORM leaves
+        unnamed; enforcement depends on ordered columns, targets and options.
+        Both sides use the same dialect/version, avoiding formatting guesses.
+    """
+    inspector = inspect(engine)
+    schema: dict[str, dict[str, object]] = {}
+    for table in inspector.get_table_names():
+        if table == "alembic_version":
+            continue
+        columns = {
+            column["name"]: (
+                str(column["type"]),
+                bool(column["nullable"]),
+                (
+                    cast(str, column["default"])[1:-1]
+                    if engine.dialect.name == "sqlite"
+                    and isinstance(column["type"], Boolean)
+                    and column.get("default") in {"'0'", "'1'"}
+                    else column.get("default")
+                ),
+                _freeze(column.get("computed")),
+                _freeze(column.get("identity")),
+            )
+            for column in inspector.get_columns(table)
+        }
+        # Retain every reflected index field: expressions, predicates, column
+        # sorting, include columns and unique flags all affect index behavior.
+        # A skipped expression index is incomplete evidence, not convergence.
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", SAWarning)
+            indexes = {_freeze(index) for index in inspector.get_indexes(table)}
+        foreign_keys = {
+            (
+                tuple(fk["constrained_columns"]),
+                fk.get("referred_schema"),
+                fk["referred_table"],
+                tuple(fk["referred_columns"]),
+                _freeze(fk.get("options", {})),
+            )
+            for fk in inspector.get_foreign_keys(table)
+        }
+        sqlite_foreign_key_actions: object = None
+        if engine.dialect.name == "sqlite":
+            # SQLite's SQL-text parser can omit inline REFERENCES actions.
+            # PRAGMA reports the actual enforcement contract independently of
+            # DDL formatting, including composite-key order (the seq column).
+            quoted_table = engine.dialect.identifier_preparer.quote_identifier(table)
+            with engine.connect() as connection:
+                rows = connection.exec_driver_sql(f"PRAGMA foreign_key_list({quoted_table})").all()
+            sqlite_foreign_key_actions = frozenset(tuple(row[1:]) for row in rows)
+        schema[table] = {
+            "columns": columns,
+            "primary_key": tuple(inspector.get_pk_constraint(table)["constrained_columns"]),
+            "checks": {_freeze(check) for check in inspector.get_check_constraints(table)},
+            "unique_constraints": {_freeze(unique) for unique in inspector.get_unique_constraints(table)},
+            "indexes": indexes,
+            "sqlite_index_definitions": (
+                _sqlite_index_definitions(engine, table) if engine.dialect.name == "sqlite" else None
+            ),
+            "foreign_keys": foreign_keys,
+            "sqlite_foreign_key_actions": sqlite_foreign_key_actions,
+        }
+    return schema
diff --git a/backend/storage/verification_database.py b/backend/storage/verification_database.py
new file mode 100644
index 0000000..c0c93f7
--- /dev/null
+++ b/backend/storage/verification_database.py
@@ -0,0 +1,88 @@
+"""Provision verification databases without touching application schemas."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from uuid import uuid4
+
+from sqlalchemy import create_engine
+from sqlalchemy.engine import make_url
+from sqlalchemy.exc import ArgumentError
+from sqlalchemy.pool import NullPool
+
+
+@contextmanager
+def disposable_postgres_database(admin_url: str) -> Iterator[str]:
+    """Create and finally drop a randomly named PostgreSQL database.
+
+    Args:
+        admin_url: Explicit test-server URL whose role can CREATE DATABASE.
+            Its path database is only a maintenance connection target. A query
+            ``dbname`` override is refused; TLS and other driver options remain.
+
+    Yields:
+        A URL for the newly created private database, using the pinned driver.
+
+    Raises:
+        ValueError: The URL cannot be parsed, does not name PostgreSQL, or has a
+            query option that would override the owned scratch database name.
+        RuntimeError: Engine setup, provisioning or cleanup fails; configuration
+            and credentials are suppressed from the public exception message.
+
+    Beginner note:
+        A shared ``public`` schema is never disposable. The only DROP target is
+        the locally generated UUID name whose CREATE succeeded in this context.
+        NullPool and AUTOCOMMIT are necessary: PostgreSQL CREATE/DROP DATABASE
+        cannot run in a transaction, and pooled scratch connections block DROP.
+        FORCE terminates only connections to our own scratch database on supported PostgreSQL.
+    """
+    # URL parsing and engine construction can fail before engine.connect. Keep
+    # those phases inside the same safe-error policy: parser/plugin errors may
+    # otherwise repeat opaque values from the supplied connection setting.
+    try:
+        url = make_url(admin_url)
+    except (ArgumentError, TypeError, ValueError):
+        raise ValueError("Verification provisioning requires a valid PostgreSQL URL") from None
+    if url.get_backend_name() != "postgresql":
+        raise ValueError("Verification provisioning requires a PostgreSQL URL")
+    # SQLAlchemy applies query parameters after translating the URL path into
+    # psycopg's dbname argument. Keeping a query override would route our yielded
+    # scratch URLs back into an existing database despite URL.set(database=...).
+    # Refuse it before constructing an engine or issuing any provisioning DDL.
+    if "dbname" in url.query:
+        raise ValueError("Verification URLs cannot contain a database-routing query option")
+    url = url.set(drivername="postgresql+psycopg")
+    # No caller-controlled identifiers enter these DDL strings. Even a hostile
+    # database component in admin_url cannot change the generated DROP target.
+    database_name = "scanner_verify_" + uuid4().hex
+    try:
+        engine = create_engine(
+            url, isolation_level="AUTOCOMMIT", poolclass=NullPool,
+            connect_args={"connect_timeout": 5},
+        )
+    except Exception:
+        raise RuntimeError("Cannot configure the disposable PostgreSQL verification connection") from None
+    created = False
+    try:
+        try:
+            with engine.connect() as connection:
+                connection.exec_driver_sql(f'CREATE DATABASE "{database_name}" TEMPLATE template0')
+                # CREATE is already committed. Track ownership before connection
+                # cleanup can raise so finally still drops our private database.
+                created = True
+        except Exception:
+            raise RuntimeError(
+                "Cannot provision disposable PostgreSQL database; check the explicit test URL and CREATEDB privilege"
+            ) from None
+        yield url.set(database=database_name).render_as_string(hide_password=False)
+    finally:
+        try:
+            if created:
+                try:
+                    with engine.connect() as connection:
+                        connection.exec_driver_sql(f'DROP DATABASE "{database_name}" WITH (FORCE)')
+                except Exception:
+                    raise RuntimeError(f"Cannot clean up disposable PostgreSQL database {database_name}") from None
+        finally:
+            engine.dispose()
diff --git a/backend/storage/verification_smoke.py b/backend/storage/verification_smoke.py
new file mode 100644
index 0000000..724131e
--- /dev/null
+++ b/backend/storage/verification_smoke.py
@@ -0,0 +1,323 @@
+"""Provider-free migration/repository verification inside the application image.
+
+Beginner note:
+    This is an explicit verification command, not an application startup hook.
+    It requires a separate PostgreSQL provisioning URL and owns disposable
+    databases and a private DATA_DIR; it never scans or sends notifications.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import json
+import os
+import subprocess  # nosec B404
+import sys
+from collections.abc import Iterator
+from contextlib import contextmanager
+from decimal import Decimal
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from alembic import command
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+from sqlalchemy import delete, func, inspect, select, text
+from sqlalchemy.engine import Engine
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from backend.storage.schema_verification import reflect_schema
+from backend.storage.verification_database import disposable_postgres_database
+
+
+@contextmanager
+def _migration_url(url: str) -> Iterator[None]:
+    """Temporarily direct Alembic's environment reader at a private database.
+
+    Args:
+        url: Caller-provisioned scratch database URL for this synchronous block.
+
+    Yields:
+        None while ``DATABASE_URL`` names the scratch database.
+
+    Beginner note:
+        Alembic reads DATABASE_URL at invocation time. Restore the original
+        value even on failure so an integration test cannot redirect later tests.
+        This helper is synchronous; callers must not run it concurrently.
+    """
+    original = os.environ.get("DATABASE_URL")
+    os.environ["DATABASE_URL"] = url
+    try:
+        yield
+    finally:
+        if original is None:
+            os.environ.pop("DATABASE_URL", None)
+        else:
+            os.environ["DATABASE_URL"] = original
+
+
+def migration_config() -> Config:
+    """Return a config anchored to the installed application migration directory.
+
+    Returns:
+        An Alembic configuration that locates the packaged migration scripts.
+        The database URL is supplied separately by ``_migration_url``.
+
+    Beginner note:
+        A programmatic Config preserves the application's logging/redaction
+        setup. Resolving from __file__ also works from a non-repository cwd.
+    """
+    config = Config()
+    config.set_main_option("script_location", str(Path(__file__).resolve().parents[2] / "migrations"))
+    return config
+
+
+def _seed_history(engine: Engine) -> tuple[int, int]:
+    """Insert initial-revision rows using only columns that existed then.
+
+    Args:
+        engine: Caller-owned scratch engine upgraded to ``20260604scan002``.
+
+    Returns:
+        The committed historical scan-run and result IDs, in that order.
+
+    Raises:
+        sqlalchemy.exc.SQLAlchemyError: The historical insert cannot complete;
+            the engine transaction rolls back both fixture rows together.
+
+    Beginner note:
+        Current ORM INSERTs cannot seed an old schema: they mention columns
+        which do not exist yet. Bound SQL uses the real historical shape and
+        lets both dialects generate IDs so sequence state remains realistic.
+    """
+    with engine.begin() as connection:
+        run_id: int = connection.execute(text(
+            "INSERT INTO scan_runs (started_at, finished_at, status, screener_key, universe_key) "
+            "VALUES (:started, :started, 'success', 'verification_legacy', 'nifty_100') RETURNING id"
+        ), {"started": dt.datetime(2026, 6, 4, tzinfo=dt.UTC).isoformat()}).scalar_one()
+        result_id: int = connection.execute(text(
+            "INSERT INTO scan_results (run_id, symbol, close_price, raw_result_json, created_at) "
+            "VALUES (:run_id, 'LEGACY', 123.4500, :payload, :created) RETURNING id"
+        ), {"run_id": run_id, "payload": '{"symbol":"LEGACY","history_marker":"retain"}',
+            "created": dt.datetime(2026, 6, 4, tzinfo=dt.UTC).isoformat()}).scalar_one()
+    return int(run_id), int(result_id)
+
+
+def _verify_repository(engine: Engine, legacy_run_id: int, legacy_result_id: int) -> None:
+    """Prove durable repository writes, rollback, uniqueness and database cascade.
+
+    Args:
+        engine: Caller-owned private engine already migrated to the current head.
+        legacy_run_id: Historical run identity returned by ``_seed_history``.
+        legacy_result_id: Historical result identity returned by that helper.
+
+    Raises:
+        AssertionError: A persisted value or database-enforced invariant differs
+            from the expected history, transaction, uniqueness, or cascade behavior.
+        sqlalchemy.exc.SQLAlchemyError: An unexpected database operation fails.
+
+    Beginner note:
+        Separate transactions ensure reads observe committed storage rather than
+        a Session identity map. A Core parent DELETE bypasses ORM cascades, so
+        missing database ON DELETE CASCADE really fails this check. All queries
+        remain inside storage; the smoke command owns the transaction lifetime.
+    """
+    from backend.storage import repository
+    from backend.storage.models import ScanResult, ScanRun, ScanStatus, UserRole
+
+    with Session(engine) as session:
+        legacy = repository.get_scan_run(session, legacy_run_id)
+        if legacy is None or legacy.status != ScanStatus.SUCCESS:
+            raise AssertionError("Historical scan header was lost or changed")
+        rows = repository.get_scan_results(session, legacy_run_id)
+        if len(rows) != 1 or rows[0].id != legacy_result_id:
+            raise AssertionError("Historical result identity was lost")
+        payload = rows[0].raw_result_json
+        if rows[0].close_price != Decimal("123.4500") or not payload or payload.get("history_marker") != "retain":
+            raise AssertionError("Historical price or JSON evidence was changed")
+    with Session(engine) as session, session.begin():
+        run = repository.create_scan_run(session, screener_key="verification_current", universe_key="nifty_100")
+        run_id = run.id
+        repository.save_scan_results(session, run, [{"symbol": "SMOKE", "close": Decimal("456.7800")}])
+        repository.finish_scan_run(session, run, status=ScanStatus.SUCCESS)
+        repository.set_user_role(session, email="smoke@example.invalid", role="admin", assigned_by=None)
+    with Session(engine) as session:
+        stored = repository.get_scan_run(session, run_id)
+        if stored is None or stored.status != ScanStatus.SUCCESS:
+            raise AssertionError("Repository write/read failed")
+        if repository.get_scan_results(session, run_id)[0].close_price != Decimal("456.7800"):
+            raise AssertionError("Repository price precision changed")
+    # Direct duplicate INSERT proves actual database uniqueness instead of
+    # relying on the repository's friendly upsert to avoid a duplicate itself.
+    with Session(engine) as session:
+        session.add(UserRole(email="smoke@example.invalid", role="admin", assigned_by=None))
+        try:
+            session.commit()
+        except IntegrityError:
+            session.rollback()
+        else:
+            raise AssertionError("Database accepted a duplicate user-role identity")
+    # A deliberate exception proves caller-owned rollback, with readback from
+    # a fresh transaction; failed writes must not create phantom scan history.
+    with Session(engine) as session:
+        rolled_back = repository.create_scan_run(
+            session, screener_key="verification_rollback", universe_key="nifty_100"
+        )
+        rolled_back_id = rolled_back.id
+        session.rollback()
+    with Session(engine) as session, session.begin():
+        if repository.get_scan_run(session, rolled_back_id) is not None:
+            raise AssertionError("Rolled-back scan survived")
+        session.execute(delete(ScanRun).where(ScanRun.id == run_id))
+        if session.scalar(select(func.count()).select_from(ScanResult).where(ScanResult.run_id == run_id)) != 0:
+            raise AssertionError("Database did not cascade result deletion")
+        if repository.get_scan_run(session, legacy_run_id) is None:
+            raise AssertionError("Deleting the scratch run deleted unrelated history")
+        if repository.delete_user_role(session, "smoke@example.invalid") != "admin":
+            raise AssertionError("Repository delete failed")
+
+
+def verify_database_pair(migrated_url: str, metadata_url: str) -> dict[str, bool]:
+    """Upgrade seeded history and compare with a separate ORM-created database.
+
+    Args:
+        migrated_url: Empty private database to build through the migration chain.
+        metadata_url: Different empty private database to build from ORM metadata.
+
+    Returns:
+        Three proof flags only after every assertion succeeds.
+
+    Raises:
+        AssertionError: Database identity, history, repository, or schema differs.
+        alembic.util.exc.CommandError: The migration chain cannot be loaded or applied.
+        sqlalchemy.exc.SQLAlchemyError: Database access or schema reflection fails.
+        sqlalchemy.exc.SAWarning: Reflection cannot describe an index completely.
+
+    Beginner note:
+        Refuse nonempty databases BEFORE any DDL. Provisioning and eventual
+        deletion belong to the caller, and this function never drops a schema.
+        Comparison is within one dialect, not a lossy SQLite/PostgreSQL mapping.
+    """
+    from backend.storage.database import _make_engine
+    from backend.storage.models import Base
+
+    if migrated_url == metadata_url:
+        raise AssertionError("Verification requires two different private databases")
+    migrated, metadata = _make_engine(migrated_url), _make_engine(metadata_url)
+    try:
+        if migrated.dialect.name != metadata.dialect.name:
+            raise AssertionError("Verification databases must use the same dialect")
+        if inspect(migrated).get_table_names() or inspect(metadata).get_table_names():
+            raise AssertionError("Verification databases must be empty")
+        config = migration_config()
+        if len(ScriptDirectory.from_config(config).get_heads()) != 1:
+            raise AssertionError("The migration chain must have exactly one head")
+        with _migration_url(migrated_url):
+            command.upgrade(config, "20260604scan002")
+            legacy_run_id, legacy_result_id = _seed_history(migrated)
+            command.upgrade(config, "head")
+        _verify_repository(migrated, legacy_run_id, legacy_result_id)
+        Base.metadata.create_all(metadata)
+        actual, expected = reflect_schema(migrated), reflect_schema(metadata)
+        differences = [
+            f"{table}.{part}"
+            for table in actual.keys() | expected.keys()
+            for part in actual.get(table, {}).keys() | expected.get(table, {}).keys()
+            if actual.get(table, {}).get(part) != expected.get(table, {}).get(part)
+        ]
+        if differences:
+            raise AssertionError("Migration/ORM schema drift: " + ", ".join(sorted(differences)))
+        return {"history_preserved": True, "repository_round_trip": True, "schema_converged": True}
+    finally:
+        migrated.dispose()
+        metadata.dispose()
+
+
+def verify_relocated_universe_sources() -> None:
+    """Read packaged source universes after moving mutable data to an empty dir.
+
+    Raises:
+        AssertionError: A curated source follows mutable ``DATA_DIR`` or does
+            not yield any symbols through the application's CSV loader.
+
+    Beginner note:
+        PR116 separates curated source CSVs from generated DATA_DIR files. This
+        check must read the package's sources even when the private runtime
+        directory contains no universes. It performs no downloads or mappings.
+    """
+    from backend.config import DATA_DIR
+    from backend.universe_builder import HEMANT_SOURCE_FILES, load_symbol_list_csv
+
+    for source in HEMANT_SOURCE_FILES.values():
+        if source.resolve().is_relative_to(DATA_DIR.resolve()):
+            raise AssertionError("Curated universe source incorrectly follows relocated DATA_DIR")
+        if not load_symbol_list_csv(source):
+            raise AssertionError("Packaged curated universe source is empty")
+
+
+def run_image_verification() -> None:
+    """Verify image sources/storage after the launcher sets private environment.
+
+    Raises:
+        SystemExit: Verification or scratch-database cleanup fails. Fixed
+            assertion diagnostics remain readable; unexpected exception messages
+            are replaced with their type so database credentials cannot leak.
+
+    Beginner note:
+        Run in the launcher's fresh process so import-time path constants read
+        the relocated DATA_DIR. Suppress driver errors to avoid echoing a DSN.
+        Assertion messages contain only our own fixed verification diagnostics.
+    """
+    try:
+        verify_relocated_universe_sources()
+        admin_url = os.environ["SCANNER_TEST_POSTGRES_ADMIN_URL"]
+        with (
+            disposable_postgres_database(admin_url) as migrated,
+            disposable_postgres_database(admin_url) as metadata,
+        ):
+            result = verify_database_pair(migrated, metadata)
+        print(json.dumps({**result, "relocated_universe_sources": True}, sort_keys=True))
+    except AssertionError as exc:
+        raise SystemExit(str(exc)) from None
+    except Exception as exc:
+        raise SystemExit(f"Application image verification failed ({type(exc).__name__})") from None
+
+
+def main() -> int:
+    """Run image verification using explicitly authorized scratch resources.
+
+    Returns:
+        Zero after the isolated child completes all source and storage checks.
+
+    Raises:
+        SystemExit: The explicit test-server setting is absent or the child fails.
+        subprocess.TimeoutExpired: The child exceeds the 180-second smoke budget.
+
+    Beginner note:
+        The package creates its normal lazy engine at import time but does not
+        connect to it. The child replaces DATABASE_URL with harmless SQLite
+        before config import; PostgreSQL writes use only generated databases.
+        Failure messages omit connection URLs and raw database exceptions.
+    """
+    admin_url = os.environ.get("SCANNER_TEST_POSTGRES_ADMIN_URL", "").strip()
+    if not admin_url:
+        raise SystemExit("SCANNER_TEST_POSTGRES_ADMIN_URL is required")
+    # backend.storage imports config before module execution. A fresh child
+    # receives DATA_DIR before that import, matching real image startup.
+    with TemporaryDirectory(prefix="scanner-image-verification-") as data_dir:
+        env = os.environ.copy()
+        env.update(DATABASE_URL="sqlite://", DATA_DIR=data_dir, ALERT_ENABLED="false")
+        # Fixed argv uses this interpreter without a shell; URLs stay in env.
+        completed = subprocess.run(  # nosec B603
+            [sys.executable, "-c", "from backend.storage.verification_smoke import run_image_verification; "
+             "run_image_verification()"],
+            env=env, check=False, timeout=180,
+        )
+        if completed.returncode:
+            raise SystemExit(completed.returncode)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/docs/architecture/components/deployment-runtime.md b/docs/architecture/components/deployment-runtime.md
index ea82b0f..1ef83e5 100644
--- a/docs/architecture/components/deployment-runtime.md
+++ b/docs/architecture/components/deployment-runtime.md
@@ -186,3 +186,38 @@ universe CSVs are per-service: persistent on the web disk, ephemeral on the cron
 
 The operator runbook (first deploy, secret file, seeding universes) is in
 [operations.md → "Deploying to Render"](../../operations.md#deploying-to-render-managed).
+
+## Application storage smoke in the final image
+
+The quality workflow provisions a PostgreSQL service matching the Compose image for each Python matrix
+worker and requires the PostgreSQL fixture lane. The Docker job still verifies
+image build and Compose health, then runs:
+
+```text
+docker compose run --rm -T --no-deps -e SCANNER_TEST_POSTGRES_ADMIN_URL -e DATABASE_URL=sqlite:// -e ALERT_ENABLED=false scanner-ui python -m backend.storage.verification_smoke
+```
+
+`SCANNER_TEST_POSTGRES_ADMIN_URL` must already be loaded from a protected
+**test-server** environment; CI supplies only its disposable Compose template
+credentials. Do not point it at a production server. The helper provisions
+unique databases, leaving the configured application database untouched. No
+market data, AI provider or alert delivery is invoked. The image needs only
+runtime dependencies; it does not install pytest/dev tools for the smoke.
+
+Beginner note: Streamlit's healthy HTTP endpoint proves that its web process is
+alive, not that migrations or repository writes work. This command upgrades
+seeded historical data, compares migration/ORM schemas and exercises repository
+transactions, uniqueness and database cascade inside the actual image. It also
+reads packaged curated universe CSVs with an empty relocated DATA_DIR, protecting
+the separation of immutable PR116 source lists from generated universe files.
+The launcher starts a fresh Python child with DATA_DIR set before imports so
+module-level path constants cannot keep the parent's original /data value.
+The private directory is removed after the child exits. The PostgreSQL contexts
+remove their databases on ordinary success or exception; CI's always-run Compose
+teardown removes the entire server/volumes even if a process is forcibly killed.
+
+A zero exit prints a JSON proof receipt. Any missing opt-in, source packaging
+failure, migration error, schema drift or repository violation fails the step.
+Database URLs/driver exception text are not printed. Local machines without
+Docker or an explicitly configured disposable PostgreSQL server cannot claim
+these lanes passed; hosted CI evidence is required before release.
diff --git a/docs/architecture/components/storage-persistence.md b/docs/architecture/components/storage-persistence.md
index bd59e0d..7175e57 100644
--- a/docs/architecture/components/storage-persistence.md
+++ b/docs/architecture/components/storage-persistence.md
@@ -247,3 +247,71 @@ per-signal schema. The PROV-003 ledger accepts richer hashed evidence through
 new durable identity or query key requires a real schema review. Any schema
 change requires an Alembic migration, matching `models.py` metadata, repository
 helpers, and updates to the owning scan-history or IPO design document.
+
+## Migration verification and disposable database contract
+
+The read-only comparator in `backend/storage/schema_verification.py` compares
+migration-built and ORM-built databases on the **same dialect**. It retains
+column types, nullability, server defaults, generated/identity definitions,
+ordered primary-key columns, checks, unique constraints, index definitions
+(including ordered columns and predicates), and foreign-key targets/options.
+`alembic_version` is bookkeeping and excluded; the existing hardcoded table
+inventory and exactly-one-Alembic-head guard remain separate checks.
+
+Beginner note: an index name can survive while its columns or WHERE condition
+change. The drift tests deliberately hold names/types constant and change one
+real enforcement rule. String-literal whitespace and case stay significant.
+Unsupported reflection (for example SQLite expression indexes) fails rather
+than silently omitting that index from the evidence.
+The only default normalization is SQLite BOOLEAN `'0'`/`'1'` versus `0`/`1`,
+whose numeric-affinity equivalence is proved by inserting and reading values.
+Foreign-key and primary-key names are not enforcement semantics: SQLite batch
+migrations name some FKs that ORM metadata leaves unnamed. Ordered columns,
+referred schema/table/columns and every reflected option are still compared.
+SQLite PRAGMA also captures actual FK update/delete actions because reflection
+of inline REFERENCES syntax can omit those options. `index_list` and
+`index_xinfo` retain collations and sort directions for explicit indexes and
+the implicit indexes backing UNIQUE/PRIMARY KEY constraints. Generated index
+names are excluded because their suffixes depend on declaration order.
+The collation regression inserts case-distinct values: one schema accepts them
+and the other rejects them, proving an enforcement difference beyond DDL text.
+
+`tests/conftest.py` owns opt-in PostgreSQL fixtures. Set
+`SCANNER_TEST_POSTGRES_ADMIN_URL` to a **test server's** maintenance database URL
+(using a role with CREATEDB), stored in a protected environment file rather
+than shell arguments. The fixture creates two `scanner_verify_<UUID>` databases
+through the existing pinned psycopg driver. It never clears a shared public
+schema or uses the application's DATABASE_URL as a fallback. Both databases
+are dropped in finally blocks; a failed second creation also cleans up the
+first. A query `dbname` option is rejected before engine construction because
+the driver would otherwise let it override the generated scratch database.
+Put the maintenance database in the URL path; TLS options remain supported.
+Connection/provisioning failures are failures once the URL is supplied.
+No URL means a local skip, unless `SCANNER_REQUIRE_POSTGRES_TESTS=1` requires the
+lane; CI sets both variables so it cannot silently skip.
+
+Run the focused contract with:
+
+```text
+python -m pytest tests/test_schema_verification.py tests/test_database_verification_smoke.py tests/test_scan_storage_migrations.py -q
+```
+
+The shared smoke scenario upgrades the initial scan-history revision, inserts
+historical rows using only its original columns, upgrades to the current head,
+and checks that IDs, lifecycle status, Decimal price and JSON evidence survive.
+It then uses current repository helpers for write/read/delete, checks a real
+database uniqueness violation and rollback, and issues a parent DELETE that
+bypasses ORM cascade so missing database cascade cannot be masked. Unrelated
+historical rows must remain. A second database built from ORM metadata provides
+the comparison reference. Existing nonempty verification targets are refused
+before DDL. This is a verification tool, not a runtime schema repair routine.
+
+`tests/test_ipo_postgres_concurrency.py` uses a separate migrated database per
+scenario to exercise the IPO state contract with real PostgreSQL sessions:
+identical concurrent calculations publish one complete pair, a paused older
+calculation cannot supersede newer evidence, and an outer rollback removes a
+staged evaluation together with its input revision. Barriers and events establish
+the intended ordering; bounded waits and database timeouts limit failed tests.
+An independent reader checks committed state while a writer remains open.
+SQLite tests of the same scenarios are useful local checks, but they cannot
+establish PostgreSQL row-lock behavior. The required hosted lane must run them.
diff --git a/tests/conftest.py b/tests/conftest.py
index 3fa54b8..028635f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -105,3 +105,84 @@ def file_db_engine(tmp_path) -> Iterator[Engine]:
 def file_session_factory(file_db_engine: Engine) -> SessionFactory:
     """A transactional session factory bound to the file-backed engine."""
     return _session_scope_factory(file_db_engine)
+
+
+@pytest.fixture
+def postgres_admin_url() -> str:
+    """Require opt-in to a PostgreSQL test server, never an application database.
+
+    Returns:
+        Explicit provisioning URL for a role with CREATEDB permission.
+
+    Beginner note:
+        Absence is an honest local skip. Once configured, every connection error
+        fails the tests. CI sets this variable and the separate required flag so
+        an accidentally removed URL cannot turn its PostgreSQL lane green.
+    """
+    import os
+
+    value = os.environ.get("SCANNER_TEST_POSTGRES_ADMIN_URL", "").strip()
+    if not value:
+        if os.environ.get("SCANNER_REQUIRE_POSTGRES_TESTS") == "1":
+            pytest.fail("SCANNER_TEST_POSTGRES_ADMIN_URL is required for this test lane")
+        pytest.skip("No disposable PostgreSQL test server configured")
+    return value
+
+
+@pytest.fixture
+def postgres_database_pair(postgres_admin_url: str) -> Iterator[tuple[str, str]]:
+    """Own two private databases for migrated-versus-ORM schema comparison.
+
+    Yields:
+        URLs for the migration database and the ORM reference database.
+
+    Beginner note:
+        Nested cleanup drops the first database even if provisioning the second
+        fails. Neither fixture rebuilds a shared schema or uses DATABASE_URL.
+    """
+    from backend.storage.verification_database import disposable_postgres_database
+
+    with (
+        disposable_postgres_database(postgres_admin_url) as migrated,
+        disposable_postgres_database(postgres_admin_url) as metadata,
+    ):
+        yield migrated, metadata
+
+
+@pytest.fixture
+def postgres_session_factory(postgres_admin_url: str) -> Iterator[SessionFactory]:
+    """Yield independent transactions on one private, fully migrated database.
+
+    Args:
+        postgres_admin_url: Explicit test-server provisioning URL, never DATABASE_URL.
+
+    Yields:
+        Commit-on-success session scopes using distinct pooled connections.
+
+    Beginner note:
+        Migrate synchronously before any worker starts: Alembic temporarily reads
+        DATABASE_URL, which is process-wide. Real PostgreSQL transactions then
+        exercise row locking rather than SQLite's database-wide writer lock.
+        Server-side timeouts bound broken locks even if a worker cannot finish;
+        dispose all pooled connections before dropping the UUID-owned database.
+    """
+    from alembic import command
+
+    from backend.storage.verification_database import disposable_postgres_database
+    from backend.storage.verification_smoke import _migration_url, migration_config
+
+    with disposable_postgres_database(postgres_admin_url) as database_url:
+        with _migration_url(database_url):
+            command.upgrade(migration_config(), "head")
+        engine = create_engine(
+            database_url,
+            future=True,
+            connect_args={
+                "connect_timeout": 5,
+                "options": "-c statement_timeout=15000 -c lock_timeout=10000",
+            },
+        )
+        try:
+            yield _session_scope_factory(engine)
+        finally:
+            engine.dispose()
diff --git a/tests/test_database_verification_smoke.py b/tests/test_database_verification_smoke.py
new file mode 100644
index 0000000..865a0ab
--- /dev/null
+++ b/tests/test_database_verification_smoke.py
@@ -0,0 +1,347 @@
+"""Safety and behavior regressions for disposable database verification."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+from sqlalchemy import inspect
+
+
+def test_repository_smoke_upgrades_seeded_history(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Execute migrations and the repository scenario on private SQLite files.
+
+    Beginner note:
+        Empty-database migration success does not prove historical rows survive
+        an upgrade, nor that the current repository can use the migrated schema.
+    """
+    from backend.storage.verification_smoke import verify_database_pair
+
+    monkeypatch.setenv("DATABASE_URL", "sqlite://")
+    result = verify_database_pair(
+        f"sqlite:///{tmp_path / 'migrated.db'}", f"sqlite:///{tmp_path / 'metadata.db'}"
+    )
+    assert result == {"history_preserved": True, "repository_round_trip": True, "schema_converged": True}
+
+
+def test_disposable_postgres_rejects_other_dialects() -> None:
+    """A test-server setting cannot accidentally authorize SQLite/application DDL.
+
+    Beginner note:
+        Only a specifically configured PostgreSQL provisioning connection may
+        create a scratch database; no application URL is an implicit fallback.
+    """
+    from backend.storage.verification_database import disposable_postgres_database
+
+    with pytest.raises(ValueError, match="PostgreSQL"), disposable_postgres_database("sqlite://"):
+        pytest.fail("Unsupported provisioning URL was accepted")
+
+
+def test_disposable_postgres_hides_malformed_url_components() -> None:
+    """A malformed provisioning URL must not echo its opaque contents.
+
+    Beginner note:
+        URL parsing used to happen before the safe-error boundary. A malformed
+        port was repeated verbatim by ValueError even though the helper promises
+        secret-safe failures. This fails before any connection can be attempted.
+    """
+    from backend.storage.verification_database import disposable_postgres_database
+
+    marker = "PRIVATE_CONFIGURATION_VALUE"
+    with pytest.raises(ValueError) as error, disposable_postgres_database(
+        f"postgresql://fixture:fixture@127.0.0.1:{marker}/postgres"
+    ):
+        pytest.fail("Malformed connection URL was accepted")
+    assert marker not in str(error.value)
+    assert "valid PostgreSQL URL" in str(error.value)
+
+
+def test_disposable_postgres_hides_engine_configuration_errors() -> None:
+    """Engine construction must use the same safe boundary as connection setup.
+
+    Beginner note:
+        An unknown SQLAlchemy plugin fails before a connection exists and echoes
+        its URL option. This fake option demonstrates the leak without reaching
+        a server; all configuration details must stay out of the public error.
+    """
+    from backend.storage.verification_database import disposable_postgres_database
+
+    marker = "PRIVATE_UNKNOWN_PLUGIN"
+    with pytest.raises(Exception) as error, disposable_postgres_database(
+        f"postgresql://fixture:fixture@127.0.0.1/postgres?plugin={marker}"
+    ):
+        pytest.fail("An unknown engine plugin was accepted")
+    assert marker not in str(error.value)
+    assert isinstance(error.value, RuntimeError)
+
+
+def test_disposable_postgres_rejects_database_override_before_engine_setup(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Reject a query database that would override both generated scratch URLs.
+
+    Beginner note:
+        SQLAlchemy lets a ``dbname`` query option override the URL path passed
+        to psycopg. Replacing only the path used to leave both supposedly private
+        URLs targeting the caller's database. Refusal must happen before CREATE
+        or any connection, and diagnostics must not repeat the supplied value.
+    """
+    from backend.storage import verification_database
+
+    created_engines = []
+
+    def unexpected_engine(*args, **kwargs):
+        """Record an unsafe early engine setup without contacting PostgreSQL."""
+        created_engines.append(True)
+        raise AssertionError("Database override reached engine construction")
+
+    monkeypatch.setattr(verification_database, "create_engine", unexpected_engine)
+    marker = "PRIVATE_EXISTING_DATABASE"
+    with pytest.raises(ValueError, match="query") as error, verification_database.disposable_postgres_database(
+        f"postgresql://fixture:fixture@127.0.0.1/postgres?dbname={marker}"
+    ):
+        pytest.fail("A database-overriding URL was accepted")
+    assert created_engines == []
+    assert marker not in str(error.value)
+
+
+def test_disposable_postgres_driver_targets_generated_database_and_keeps_tls(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Check actual driver arguments, not just the displayed scratch URL path.
+
+    Beginner note:
+        A safe-looking URL is insufficient when query options can override its
+        path. This no-network fixture runs SQLAlchemy's real psycopg argument
+        conversion and proves the generated CREATE target is also the effective
+        connection database. Legitimate TLS options must survive the boundary.
+    """
+    from contextlib import contextmanager
+    from types import SimpleNamespace
+
+    from sqlalchemy.dialects.postgresql.psycopg import PGDialect_psycopg
+    from sqlalchemy.engine import make_url
+
+    from backend.storage import verification_database
+
+    commands: list[str] = []
+
+    @contextmanager
+    def connection():
+        """Record scratch CREATE/DROP statements without opening a connection."""
+        yield SimpleNamespace(exec_driver_sql=commands.append)
+
+    monkeypatch.setattr(verification_database, "create_engine", lambda *_args, **_kwargs: SimpleNamespace(
+        connect=connection, dispose=lambda: None,
+    ))
+    with verification_database.disposable_postgres_database(
+        "postgresql://fixture:fixture@127.0.0.1/postgres?sslmode=verify-full"
+    ) as private_url:
+        created_name = commands[0].split('"')[1]
+        _args, kwargs = PGDialect_psycopg().create_connect_args(make_url(private_url))
+        assert kwargs["dbname"] == created_name
+        assert created_name.startswith("scanner_verify_")
+        assert kwargs["sslmode"] == "verify-full"
+    assert commands[1] == f'DROP DATABASE "{created_name}" WITH (FORCE)'
+
+
+def test_disposable_postgres_cleans_up_when_create_connection_exit_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Track a successful CREATE before its connection context can raise.
+
+    Beginner note:
+        Creation is committed immediately in AUTOCOMMIT. If connection cleanup
+        then fails, the private database still exists and must be dropped. This
+        fixture records DDL without reaching a server and proves the DROP target
+        is exactly the generated CREATE target, never the configured database.
+    """
+    from contextlib import contextmanager
+    from types import SimpleNamespace
+
+    from backend.storage import verification_database
+
+    commands: list[str] = []
+    disposed: list[bool] = []
+
+    @contextmanager
+    def connection():
+        """Model a close failure only after the first committed DDL statement.
+
+        Beginner note:
+            The second context is the finally cleanup path and closes normally;
+            changing the flag after the first context exits would miss that path.
+        """
+        yield SimpleNamespace(exec_driver_sql=commands.append)
+        if len(commands) == 1:
+            raise RuntimeError("Fixture close failure")
+
+    monkeypatch.setattr(verification_database, "create_engine", lambda *_args, **_kwargs: SimpleNamespace(
+        connect=connection, dispose=lambda: disposed.append(True),
+    ))
+    with pytest.raises(RuntimeError, match="Cannot provision"), verification_database.disposable_postgres_database(
+        "postgresql://fixture:fixture@127.0.0.1/existing_database"
+    ):
+        pytest.fail("A failed provisioning connection was yielded")
+    assert len(commands) == 2
+    created_name = commands[0].split('"')[1]
+    assert created_name.startswith("scanner_verify_")
+    assert commands[1] == f'DROP DATABASE "{created_name}" WITH (FORCE)'
+    assert disposed == [True]
+
+
+def test_explicit_unreachable_postgres_is_failure() -> None:
+    """A requested but unreachable server must fail rather than silently skip.
+
+    Beginner note:
+        This loopback closed-port URL never reaches a provider or application
+        database. It distinguishes a missing optional setting from broken CI.
+    """
+    from backend.storage.verification_database import disposable_postgres_database
+
+    with (
+        pytest.raises(RuntimeError, match="provision disposable PostgreSQL"),
+        disposable_postgres_database("postgresql+psycopg://test:test@127.0.0.1:1/postgres?connect_timeout=1"),
+    ):
+        pytest.fail("The intentionally unreachable server accepted a connection")
+
+
+def test_postgres_migrations_and_repository(postgres_database_pair: tuple[str, str]) -> None:
+    """Run the same history and enforcement contract against a real PG server.
+
+    Beginner note:
+        The fixture provisions TWO unique databases, never drops public tables
+        in a configured database, and fails if an explicitly requested server
+        is unavailable. CI supplies that setting, so this test cannot skip there.
+    """
+    from backend.storage.verification_smoke import verify_database_pair
+
+    assert verify_database_pair(*postgres_database_pair)["schema_converged"] is True
+
+
+def test_disposable_postgres_cleanup(postgres_admin_url: str) -> None:
+    """Exception cleanup removes only the database created by this context.
+
+    Beginner note:
+        The nested fixture outlives the scratch database so the readback can
+        prove the database was dropped even when the body raises an exception.
+    """
+    from sqlalchemy import create_engine, text
+    from sqlalchemy.engine import make_url
+
+    from backend.storage.verification_database import disposable_postgres_database
+
+    database_name = ""
+    with (
+        pytest.raises(LookupError, match="intentional fixture failure"),
+        disposable_postgres_database(postgres_admin_url) as url,
+    ):
+        database_name = make_url(url).database or ""
+        engine = create_engine(url)
+        try:
+            assert inspect(engine).get_table_names() == []
+        finally:
+            engine.dispose()
+        raise LookupError("intentional fixture failure")
+    engine = create_engine(make_url(postgres_admin_url).set(drivername="postgresql+psycopg"))
+    try:
+        with engine.connect() as connection:
+            assert connection.scalar(
+                text("SELECT count(*) FROM pg_database WHERE datname = :name"), {"name": database_name}
+            ) == 0
+    finally:
+        engine.dispose()
+
+
+def test_smoke_cli_requires_explicit_provisioning_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The image CLI must not borrow DATABASE_URL if its test setting is absent.
+
+    Beginner note:
+        Operators may run a smoke command from an application container. Merely
+        having production credentials in that container cannot authorize writes.
+    """
+    import os
+
+    env = os.environ.copy()
+    env.pop("SCANNER_TEST_POSTGRES_ADMIN_URL", None)
+    env["DATABASE_URL"] = "sqlite:///DO_NOT_CREATE_SMOKE.db"
+    completed = subprocess.run(
+        [sys.executable, "-m", "backend.storage.verification_smoke"],
+        env=env, capture_output=True, text=True, check=False, timeout=30,
+    )
+    assert completed.returncode != 0
+    assert "SCANNER_TEST_POSTGRES_ADMIN_URL is required" in completed.stderr
+    assert not Path("DO_NOT_CREATE_SMOKE.db").exists()
+
+
+def test_required_postgres_lane_cannot_silently_skip() -> None:
+    """Removing the CI URL must be a failing fixture, not a green skipped lane.
+
+    Beginner note:
+        This runs only the single PostgreSQL case in a child with no URL, so it
+        cannot contact a server. The child must fail before provisioning begins.
+    """
+    import os
+
+    env = os.environ.copy()
+    env.pop("SCANNER_TEST_POSTGRES_ADMIN_URL", None)
+    env.update(DATABASE_URL="sqlite://", SCANNER_REQUIRE_POSTGRES_TESTS="1")
+    result = subprocess.run(
+        [sys.executable, "-m", "pytest", "tests/test_database_verification_smoke.py",
+         "-k", "test_postgres_migrations_and_repository", "-q", "--tb=short"],
+        env=env, capture_output=True, text=True, check=False, timeout=60,
+    )
+    assert result.returncode == 1
+    assert "SCANNER_TEST_POSTGRES_ADMIN_URL is required for this test lane" in result.stdout
+    assert "1 skipped" not in result.stdout
+
+
+def test_nonempty_database_is_refused_before_migration(tmp_path: Path) -> None:
+    """An accidentally reused target must retain its existing table and rows.
+
+    Beginner note:
+        Provisioning contexts normally guarantee emptiness. This independent
+        guard ensures direct helper callers cannot rewrite an existing schema.
+    """
+    from sqlalchemy import create_engine, text
+
+    from backend.storage.verification_smoke import verify_database_pair
+
+    url = f"sqlite:///{tmp_path / 'occupied.db'}"
+    engine = create_engine(url)
+    try:
+        with engine.begin() as connection:
+            connection.execute(text("CREATE TABLE sentinel (marker TEXT)"))
+            connection.execute(text("INSERT INTO sentinel VALUES ('retain')"))
+        with pytest.raises(AssertionError, match="must be empty"):
+            verify_database_pair(url, f"sqlite:///{tmp_path / 'reference.db'}")
+        with engine.connect() as connection:
+            assert connection.scalar(text("SELECT marker FROM sentinel")) == "retain"
+        assert inspect(engine).get_table_names() == ["sentinel"]
+    finally:
+        engine.dispose()
+
+
+@pytest.mark.parametrize("source_in_data_dir", [False, True])
+def test_image_sources_are_read_only_and_outside_runtime_data(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, source_in_data_dir: bool
+) -> None:
+    """Read a packaged CSV and reject a source accidentally moved into DATA_DIR.
+
+    Beginner note:
+        This fixture uses a real CSV parser without fetching an instrument
+        master. Distinct source/runtime roots are the important image contract;
+        the current product source-path mapping is checked in hosted smoke.
+    """
+    from backend import config, universe_builder
+    from backend.storage.verification_smoke import verify_relocated_universe_sources
+
+    data_dir = tmp_path / "runtime"
+    data_dir.mkdir()
+    source = (data_dir if source_in_data_dir else tmp_path) / "curated.csv"
+    content = "symbol\nEXAMPLE\n"
+    source.write_text(content, encoding="utf-8")
+    monkeypatch.setattr(config, "DATA_DIR", data_dir)
+    monkeypatch.setattr(universe_builder, "HEMANT_SOURCE_FILES", {"example": source})
+    if source_in_data_dir:
+        with pytest.raises(AssertionError, match="relocated DATA_DIR"):
+            verify_relocated_universe_sources()
+    else:
+        verify_relocated_universe_sources()
+    assert source.read_text(encoding="utf-8") == content
diff --git a/tests/test_ipo_postgres_concurrency.py b/tests/test_ipo_postgres_concurrency.py
new file mode 100644
index 0000000..29fdc58
--- /dev/null
+++ b/tests/test_ipo_postgres_concurrency.py
@@ -0,0 +1,239 @@
+"""IPO scoring transaction contracts against disposable migrated PostgreSQL.
+
+Beginner note:
+    SQLite's single-writer behavior cannot prove PostgreSQL row-lock semantics.
+    These tests opt in through SCANNER_TEST_POSTGRES_ADMIN_URL and use separate
+    real sessions. Synchronization pauses calculation, never replaces persistence.
+    Missing local PostgreSQL is a skip; the required CI lane treats absence as
+    failure. No shared application database or external evidence source is used.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+from collections.abc import Iterator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from decimal import Decimal
+from pathlib import Path
+from threading import Barrier, Event
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+from test_ipo_scoring_service import _AS_OF, _issue_data, _scored_issue
+
+from backend.ipo import repository
+from backend.ipo.repository import SessionFactory
+from backend.ipo.scoring import service
+from backend.ipo.scoring.state import IpoScoringSnapshot
+from backend.storage.ipo_repository import get_ipo_scoring_state_values, update_ipo_issue_row
+from backend.storage.models import IpoRecommendation, IpoScore
+
+
+def _history_counts(session: Session, issue_id: int) -> tuple[int, int]:
+    """Count both halves of persisted evaluation history independently.
+
+    Args:
+        session: Caller-owned inspection transaction.
+        issue_id: Issue whose immutable history is inspected.
+
+    Returns:
+        Score count and recommendation count, including incomplete pairs.
+
+    Beginner note:
+        The public history list joins complete pairs. Counting both tables also
+        catches an orphan insert that a successful joined read could conceal.
+    """
+    scores = session.scalar(select(func.count()).select_from(IpoScore).where(IpoScore.issue_id == issue_id))
+    recommendations = session.scalar(
+        select(func.count()).select_from(IpoRecommendation).where(
+            IpoRecommendation.score_id.in_(select(IpoScore.id).where(IpoScore.issue_id == issue_id))
+        )
+    )
+    assert scores is not None and recommendations is not None
+    return scores, recommendations
+
+
+def test_postgres_identical_concurrent_calculations_publish_one_pair(
+    postgres_session_factory: SessionFactory, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Concurrent identical jobs return the same complete, current evaluation.
+
+    Beginner note:
+        Both jobs finish their real reads before either publishes. Removing the
+        publication serialization/idempotency guard would allow duplicate pairs,
+        leak an integrity failure, or leave current state pointing at other history.
+        A barrier establishes the overlap; no elapsed-time assertion is involved.
+    """
+    issue = _scored_issue(postgres_session_factory, tmp_path)
+    snapshots_ready = Barrier(2)
+    load_snapshot = service.load_ipo_scoring_snapshot
+
+    def synchronized_snapshot(
+        issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = postgres_session_factory
+    ) -> IpoScoringSnapshot:
+        """Detach real evidence, then release both calculations together.
+
+        Args:
+            issue_id: Issue being scored.
+            as_of: Frozen business clock supplied by the scoring service.
+            session_factory: Real transaction factory forwarded unchanged.
+
+        Returns:
+            The unchanged, revision-validated evidence snapshot.
+        """
+        snapshot = load_snapshot(issue_id, as_of=as_of, session_factory=session_factory)
+        snapshots_ready.wait(timeout=15)
+        return snapshot
+
+    monkeypatch.setattr(service, "load_ipo_scoring_snapshot", synchronized_snapshot)
+    with ThreadPoolExecutor(max_workers=2) as workers:
+        futures = [
+            workers.submit(service.rescore_issue, issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+            for _ in range(2)
+        ]
+        outcomes = [future.result(timeout=30) for future in futures]
+    # Restore before the current reader, which also loads a scoring snapshot.
+    monkeypatch.setattr(service, "load_ipo_scoring_snapshot", load_snapshot)
+    assert {outcome.status for outcome in outcomes} == {"evaluated", "skipped_unchanged"}
+    evaluation = outcomes[0].evaluation
+    assert evaluation is not None and outcomes[1].evaluation == evaluation
+    with postgres_session_factory() as session:
+        assert _history_counts(session, issue.id) == (1, 1)
+    current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+    assert current.fresh and current.evaluation == evaluation
+    assert current.snapshot.state.current_score_id == evaluation.score_id
+    assert current.snapshot.state.evaluated_revision == current.snapshot.state.input_revision
+
+
+def test_postgres_older_calculation_cannot_supersede_newer_revision(
+    postgres_session_factory: SessionFactory, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An older detached calculation retries and reuses the newer committed pair.
+
+    Beginner note:
+        Pause after the old snapshot's transaction closes, then commit changed
+        pricing and its evaluation through independent sessions. If publication
+        loses its revision compare-and-set, releasing the old calculation would
+        insert obsolete history or supersede the newer current receipt.
+    """
+    issue = _scored_issue(postgres_session_factory, tmp_path)
+    old_snapshot_ready = Event()
+    newer_committed = Event()
+    load_snapshot = service.load_ipo_scoring_snapshot
+
+    def paused_factory() -> Iterator[Session]:
+        """Identify the old worker without changing transaction behavior.
+
+        Yields:
+            A real independent session, committed or rolled back by the fixture.
+
+        Beginner note:
+            The wrapper's identity limits the pause to the old worker. Its
+            transactions still come from the shared PostgreSQL engine; no lock
+            or persistence operation is mocked.
+        """
+        with postgres_session_factory() as session:
+            yield session
+
+    old_worker_factory = contextmanager(paused_factory)
+
+    def pause_old_snapshot(
+        issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = postgres_session_factory
+    ) -> IpoScoringSnapshot:
+        """Hold only the old worker's first detached evidence bundle.
+
+        Args:
+            issue_id: Issue being scored.
+            as_of: Service's frozen business time.
+            session_factory: Identifies the old worker while forwarding real I/O.
+
+        Returns:
+            An unchanged snapshot after the newer commit, or immediately on retry.
+        """
+        snapshot = load_snapshot(issue_id, as_of=as_of, session_factory=session_factory)
+        if session_factory is old_worker_factory and not old_snapshot_ready.is_set():
+            old_snapshot_ready.set()
+            assert newer_committed.wait(timeout=20), "Newer revision was never committed"
+        return snapshot
+
+    monkeypatch.setattr(service, "load_ipo_scoring_snapshot", pause_old_snapshot)
+    with ThreadPoolExecutor(max_workers=1) as workers:
+        older = workers.submit(service.rescore_issue, issue.id, as_of=_AS_OF, session_factory=old_worker_factory)
+        try:
+            assert old_snapshot_ready.wait(timeout=15), "Old worker did not capture its snapshot"
+            repository.update_issue(
+                issue.id, _issue_data(price_band_high=Decimal("300")), session_factory=postgres_session_factory
+            )
+            newer = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+            assert newer.status == "evaluated" and newer.evaluation is not None
+            committed = service.get_current_evaluation(
+                issue.id, as_of=_AS_OF, session_factory=postgres_session_factory
+            )
+        finally:
+            # Release on assertion/error too, so executor cleanup never hangs.
+            newer_committed.set()
+        retried = older.result(timeout=30)
+    assert retried.status == "skipped_unchanged" and retried.evaluation == newer.evaluation
+    assert repository.list_evaluations(issue.id, session_factory=postgres_session_factory) == [newer.evaluation]
+    current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+    assert current.fresh and current.evaluation == newer.evaluation
+    assert current.snapshot.state.input_revision == committed.snapshot.state.input_revision
+    assert current.snapshot.state.evaluated_revision == committed.snapshot.state.input_revision
+    with postgres_session_factory() as session:
+        assert _history_counts(session, issue.id) == (1, 1)
+
+
+def test_postgres_caller_rollback_restores_evidence_revision_and_selection(
+    postgres_session_factory: SessionFactory, tmp_path: Path
+) -> None:
+    """An outer caller failure rolls back input and nested score publication.
+
+    Beginner note:
+        A service may flush and use a savepoint, but the caller owns commit.
+        Independent readers must see the original receipt while writes are open,
+        and after rollback both the evidence token and immutable history must be
+        restored. This catches an internal commit or escaped savepoint insertion.
+    """
+    issue = _scored_issue(postgres_session_factory, tmp_path)
+    original = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+    assert original.evaluation is not None
+    with postgres_session_factory() as session:
+        original_state = get_ipo_scoring_state_values(session, issue.id)
+    assert original_state is not None
+
+    with pytest.raises(RuntimeError, match="caller aborts transaction"), postgres_session_factory() as writer:
+        changed = update_ipo_issue_row(writer, issue.id, {"price_band_high": Decimal("300")})
+        assert changed is not None
+
+        @contextmanager
+        def caller_session() -> Iterator[Session]:
+            """Lend the outer session while retaining the caller's commit authority.
+
+            Yields:
+                The existing caller transaction, without commit or rollback.
+
+            Beginner note:
+                Snapshot reads and publication share this transaction so the
+                final caller rollback includes evidence and the nested pair.
+            """
+            yield writer
+
+        staged = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=caller_session)
+        assert staged.status == "evaluated" and staged.evaluation is not None
+        staged_state = get_ipo_scoring_state_values(writer, issue.id)
+        assert staged_state is not None and staged_state[0] > original_state[0]
+        assert staged_state[1] == staged.evaluation.score_id and staged_state[2] == staged_state[0]
+        assert _history_counts(writer, issue.id) == (2, 2)
+        with postgres_session_factory() as reader:
+            assert get_ipo_scoring_state_values(reader, issue.id) == original_state
+            assert _history_counts(reader, issue.id) == (1, 1)
+        raise RuntimeError("caller aborts transaction")
+
+    with postgres_session_factory() as reader:
+        assert get_ipo_scoring_state_values(reader, issue.id) == original_state
+        assert _history_counts(reader, issue.id) == (1, 1)
+    current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory)
+    assert current.fresh and current.evaluation == original.evaluation
+    assert current.snapshot.inputs.issue.price_band_high == Decimal("242")
diff --git a/tests/test_scan_storage_migrations.py b/tests/test_scan_storage_migrations.py
index 08906f6..1c11e1f 100644
--- a/tests/test_scan_storage_migrations.py
+++ b/tests/test_scan_storage_migrations.py
@@ -977,34 +977,18 @@ def test_ensure_database_schema_detects_schema_drift(monkeypatch, tmp_path: Path
 
 
 def _reflect_schema(engine) -> dict[str, dict[str, object]]:
-    """Return a comparable {table: columns/indexes/foreign_keys} structure.
+    """Reflect full enforcement contracts and dispose this test-owned engine.
 
-    Alembic's bookkeeping ``alembic_version`` table is skipped so a migrated
-    database and a ``create_all`` database describe the same application schema.
-    The engine is disposed before returning.
+    Beginner note:
+        The reusable fingerprint also powers PostgreSQL and image verification;
+        the wrapper retains this module's existing engine ownership convention.
     """
-    inspector = inspect(engine)
-    schema: dict[str, dict[str, object]] = {}
-    for table in inspector.get_table_names():
-        if table == "alembic_version":
-            continue
-        columns = {
-            column["name"]: (str(column["type"]), bool(column["nullable"]))
-            for column in inspector.get_columns(table)
-        }
-        indexes = {index["name"] for index in inspector.get_indexes(table)}
-        foreign_keys = {
-            (
-                tuple(fk["constrained_columns"]),
-                fk["referred_table"],
-                tuple(fk["referred_columns"]),
-                fk.get("options", {}).get("ondelete"),
-            )
-            for fk in inspector.get_foreign_keys(table)
-        }
-        schema[table] = {"columns": columns, "indexes": indexes, "foreign_keys": foreign_keys}
-    engine.dispose()
-    return schema
+    from backend.storage.schema_verification import reflect_schema
+
+    try:
+        return reflect_schema(engine)
+    finally:
+        engine.dispose()
 
 
 def test_valid005_backfills_attempts_and_only_missing_computed_benchmarks(monkeypatch, tmp_path: Path):
diff --git a/tests/test_schema_verification.py b/tests/test_schema_verification.py
new file mode 100644
index 0000000..df37cd9
--- /dev/null
+++ b/tests/test_schema_verification.py
@@ -0,0 +1,223 @@
+"""Regression proofs that the schema fingerprint detects enforcement drift."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.exc import IntegrityError
+from test_scan_storage_migrations import _reflect_schema
+
+
+@pytest.mark.parametrize(
+    ("before", "after"),
+    [
+        ("CHECK (amount > 0)", "CHECK (amount >= 0)"),
+        ("UNIQUE (code)", "UNIQUE (amount)"),
+        ("PRIMARY KEY (id, code)", "PRIMARY KEY (code, id)"),
+    ],
+    ids=["check-expression", "unique-columns", "primary-key-order"],
+)
+def test_constraint_drift_is_visible(tmp_path: Path, before: str, after: str) -> None:
+    """Keep identical column types/names while changing a real constraint.
+
+    Beginner note:
+        The old comparator only saw columns and index names, so it could approve
+        a migration that accepts forbidden values or changes row identity.
+    """
+    schemas = []
+    for position, constraint in enumerate((before, after)):
+        engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}")
+        with engine.begin() as connection:
+            connection.exec_driver_sql(
+                f"CREATE TABLE sample (id INTEGER NOT NULL, code TEXT NOT NULL, "
+                f"amount INTEGER NOT NULL, {constraint})"
+            )
+        schemas.append(_reflect_schema(engine))
+    assert schemas[0] != schemas[1]
+
+
+@pytest.mark.parametrize(
+    ("before", "after"),
+    [
+        ("CREATE INDEX sample_idx ON sample (id, code)", "CREATE INDEX sample_idx ON sample (code, id)"),
+        ("CREATE INDEX sample_idx ON sample (code)", "CREATE UNIQUE INDEX sample_idx ON sample (code)"),
+        ("CREATE INDEX sample_idx ON sample (id) WHERE id > 0", "CREATE INDEX sample_idx ON sample (id) WHERE id >= 0"),
+    ],
+    ids=["index-column-order", "index-uniqueness", "index-predicate"],
+)
+def test_same_named_index_drift_is_visible(tmp_path: Path, before: str, after: str) -> None:
+    """Changing an index definition must matter even when its name survives.
+
+    Beginner note:
+        A partial unique index only protects rows matching its predicate. A name
+        comparison cannot prove either the uniqueness or the protected rows.
+    """
+    schemas = []
+    for position, statement in enumerate((before, after)):
+        engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}")
+        with engine.begin() as connection:
+            connection.exec_driver_sql("CREATE TABLE sample (id INTEGER, code TEXT)")
+            connection.exec_driver_sql(statement)
+        schemas.append(_reflect_schema(engine))
+    assert schemas[0] != schemas[1]
+
+
+@pytest.mark.parametrize("index_kind", ["index", "unique", "primary_key"])
+def test_sqlite_unique_index_collation_drift_is_visible(tmp_path: Path, index_kind: str) -> None:
+    """Expose different accepted values despite identical reflected index columns.
+
+    Beginner note:
+        SQLAlchemy omits SQLite index collations without a warning. Previously
+        the comparator treated BINARY and NOCASE unique indexes as equivalent,
+        even though only one accepts two differently capitalized spellings.
+        Table-level UNIQUE and PRIMARY KEY rules use implicit indexes and need
+        the same check.
+    """
+    schemas = []
+    accepted_case_variant = []
+    for position, collation in enumerate(("BINARY", "NOCASE")):
+        engine = create_engine(f"sqlite:///{tmp_path / f'collation-{position}.db'}")
+        try:
+            with engine.begin() as connection:
+                if index_kind != "index":
+                    constraint = "UNIQUE" if index_kind == "unique" else "PRIMARY KEY"
+                    connection.exec_driver_sql(
+                        f"CREATE TABLE sample (code TEXT, {constraint} (code COLLATE {collation}))"
+                    )
+                else:
+                    connection.exec_driver_sql("CREATE TABLE sample (code TEXT)")
+                    connection.exec_driver_sql(
+                        f"CREATE UNIQUE INDEX sample_idx ON sample (code COLLATE {collation})"
+                    )
+                connection.exec_driver_sql("INSERT INTO sample VALUES ('Mixed')")
+                try:
+                    connection.exec_driver_sql("INSERT INTO sample VALUES ('mixed')")
+                except IntegrityError:
+                    accepted_case_variant.append(False)
+                else:
+                    accepted_case_variant.append(True)
+            schemas.append(_reflect_schema(engine))
+        finally:
+            engine.dispose()
+    assert accepted_case_variant == [True, False]
+    assert schemas[0] != schemas[1]
+
+
+def test_sqlite_implicit_index_numbering_does_not_create_false_drift(tmp_path: Path) -> None:
+    """Compare constraint definitions independently of SQLite autoindex suffixes.
+
+    Beginner note:
+        Reordering two equivalent UNIQUE declarations changes SQLite's generated
+        index names. The new collation guard must retain both constraints while
+        ignoring those incidental names, so harmless migration formatting does
+        not become a false schema mismatch.
+    """
+    schemas = []
+    for position, constraints in enumerate(("UNIQUE (id), UNIQUE (code)", "UNIQUE (code), UNIQUE (id)")):
+        engine = create_engine(f"sqlite:///{tmp_path / f'autoindex-{position}.db'}")
+        try:
+            with engine.begin() as connection:
+                connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, code TEXT, {constraints})")
+            schemas.append(_reflect_schema(engine))
+        finally:
+            engine.dispose()
+    assert schemas[0] == schemas[1]
+
+
+def test_sqlite_index_sort_direction_drift_is_visible(tmp_path: Path) -> None:
+    """Retain descending index order that SQLite's generic reflection drops.
+
+    Beginner note:
+        Both indexes have the same name and column list, but the second key's
+        direction changes the ordering that the index supports. The fingerprint
+        must compare the actual key definition, not just the column names.
+    """
+    schemas = []
+    for position, direction in enumerate(("ASC", "DESC")):
+        engine = create_engine(f"sqlite:///{tmp_path / f'direction-{position}.db'}")
+        try:
+            with engine.begin() as connection:
+                connection.exec_driver_sql("CREATE TABLE sample (id INTEGER, code TEXT)")
+                connection.exec_driver_sql(f"CREATE INDEX sample_idx ON sample (id ASC, code {direction})")
+            schemas.append(_reflect_schema(engine))
+        finally:
+            engine.dispose()
+    assert schemas[0] != schemas[1]
+
+
+@pytest.mark.parametrize("defaults", [("0", "1"), ("'A B'", "'A  B'"), ("'Mixed'", "'mixed'")])
+def test_server_default_drift_preserves_literal_semantics(tmp_path: Path, defaults: tuple[str, str]) -> None:
+    """Defaults and significant text inside SQL quotes must stay distinguishable.
+
+    Beginner note:
+        Removing all whitespace or lowercasing SQL also changes string literals.
+        These pairs expose normalization that would hide different stored values.
+    """
+    schemas = []
+    for position, default in enumerate(defaults):
+        engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}")
+        with engine.begin() as connection:
+            connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, code TEXT DEFAULT {default})")
+        schemas.append(_reflect_schema(engine))
+    assert schemas[0] != schemas[1]
+
+
+def test_foreign_key_update_action_drift_is_visible(tmp_path: Path) -> None:
+    """Keep FK names/targets/deletion rules equal while changing update behavior.
+
+    Beginner note:
+        The previous comparator retained ON DELETE but discarded ON UPDATE;
+        that omission could approve a migration that breaks parent-key updates.
+    """
+    schemas = []
+    for position, action in enumerate(("CASCADE", "RESTRICT")):
+        engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}")
+        with engine.begin() as connection:
+            connection.exec_driver_sql("CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+            connection.exec_driver_sql(
+                "CREATE TABLE sample (parent_id INTEGER REFERENCES parent(id) "
+                f"ON DELETE CASCADE ON UPDATE {action})"
+            )
+        schemas.append(_reflect_schema(engine))
+    assert schemas[0] != schemas[1]
+
+
+@pytest.mark.parametrize("boolean", [0, 1])
+def test_sqlite_boolean_default_spellings_store_identical_values(tmp_path: Path, boolean: int) -> None:
+    """Prove the precise SQLite Boolean formatting exception before normalizing.
+
+    Beginner note:
+        The migration uses DEFAULT 0; SQLAlchemy's string default emits '0'.
+        SQLite BOOLEAN has numeric affinity, so both store integer zero. This
+        exception must never be applied to TEXT, where quoted content matters.
+    """
+    schemas = []
+    values = []
+    for position, default in enumerate((str(boolean), f"'{boolean}'")):
+        engine = create_engine(f"sqlite:///{tmp_path / f'bool-{position}.db'}")
+        with engine.begin() as connection:
+            connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, flag BOOLEAN DEFAULT {default})")
+            connection.exec_driver_sql("INSERT INTO sample (id) VALUES (1)")
+            values.append(connection.exec_driver_sql("SELECT flag, typeof(flag) FROM sample").one())
+        schemas.append(_reflect_schema(engine))
+    assert values == [(boolean, "integer"), (boolean, "integer")]
+    assert schemas[0] == schemas[1]
+
+
+def test_unreflectable_sqlite_expression_index_is_not_silently_omitted(tmp_path: Path) -> None:
+    """Unsupported reflection must fail closed instead of comparing incomplete schemas.
+
+    Beginner note:
+        SQLAlchemy warns and skips SQLite expression indexes. Ignoring that
+        warning would let a missing enforcement index evade this comparator.
+    """
+    from sqlalchemy.exc import SAWarning
+
+    engine = create_engine(f"sqlite:///{tmp_path / 'expression.db'}")
+    with engine.begin() as connection:
+        connection.exec_driver_sql("CREATE TABLE sample (code TEXT)")
+        connection.exec_driver_sql("CREATE UNIQUE INDEX ix_code ON sample (lower(code))")
+    with pytest.raises(SAWarning, match="expression-based index"):
+        _reflect_schema(engine)
diff --git a/tests/test_supply_chain_policy.py b/tests/test_supply_chain_policy.py
index 97b84ac..d2c584c 100644
--- a/tests/test_supply_chain_policy.py
+++ b/tests/test_supply_chain_policy.py
@@ -734,3 +734,38 @@ def test_screener_guide_matches_registry_chart_golden_and_ci_contracts():
     assert "series.isna().any()" not in text
     for command in CI_COMMANDS:
         assert command in text
+
+
+def test_ci_requires_real_postgres_and_application_image_smoke():
+    """Protect the executable PostgreSQL and final-image verification lanes.
+
+    Beginner note:
+        A health endpoint and pip installation cannot prove migrations or
+        repository writes work in the deployment image. Required opt-in prevents
+        missing CI environment from silently skipping PostgreSQL integration.
+        The ordinary suite also needs a private file-backed SQLite default:
+        Alembic and application sessions otherwise open different in-memory
+        databases, so a successful upgrade cannot initialize the later session.
+        GitHub resolves job-level environment before a runner exists. The
+        runner.temp expression must live on the test step, where that context
+        is available, or GitHub rejects the workflow before creating any jobs.
+    """
+    workflow = yaml.safe_load((ROOT / ".github/workflows/quality-and-security.yml").read_text(encoding="utf-8"))
+    verify = workflow["jobs"]["verify"]
+    assert "DATABASE_URL" not in verify["env"]
+    test_step = next(step for step in verify["steps"] if step["name"] == "Run tests with coverage gate")
+    assert test_step["env"]["DATABASE_URL"] == "sqlite:///${{ runner.temp }}/scanner-default.db"
+    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
+    assert verify["services"]["postgres"]["image"] == compose["services"]["postgres"]["image"]
+    assert verify["env"]["SCANNER_REQUIRE_POSTGRES_TESTS"] == "1"
+    assert "127.0.0.1:5432/postgres" in verify["env"]["SCANNER_TEST_POSTGRES_ADMIN_URL"]
+    smoke_steps = workflow["jobs"]["docker-build"]["steps"]
+    application_smoke = next(
+        step for step in smoke_steps if step["name"] == "Verify application storage in final image"
+    )
+    assert "docker compose run --rm -T --no-deps" in application_smoke["run"]
+    assert "-e SCANNER_TEST_POSTGRES_ADMIN_URL" in application_smoke["run"]
+    assert "scanner-ui python -m backend.storage.verification_smoke" in application_smoke["run"]
+    assert "@postgres:5432/postgres" in application_smoke["env"]["SCANNER_TEST_POSTGRES_ADMIN_URL"]
+    teardown = next(step for step in smoke_steps if step["name"] == "Tear down Compose stack")
+    assert teardown["if"] == "always()"