V4.6 Phase 2: schema re-level in a single atomic pass

These changes all regenerate the same schema, so they land together and revert
together. A partially applied schema pass is not a valid state.

Remove hand-rolled migrations [HIGH-05]
- Delete upgrade_schema and the _upgrade_person_family_search_id /
  _upgrade_v42_evidence_tables / _upgrade_v45_selection_columns chain, plus the
  two tests that exercised them. The DDL was SQLite-shaped raw SQL that would
  not have run on PostgreSQL. create_all now derives everything from metadata
  and remains gated by Settings.should_bootstrap_schema. No raw ALTER TABLE or
  CREATE INDEX string remains in src.

Break the foreign key cycle [HIGH-08]
- Declare Source.preferred_execution_attempt_id with use_alter=True and an
  explicit constraint name. source / job_source / execution_attempt formed an
  unresolvable cycle that made metadata.sorted_tables emit an SAWarning and
  order execution_attempt before source, which would have been a hard
  create_all failure on PostgreSQL and was invisible on SQLite.
- As a side effect the column is now a dialect-aware Uuid rather than the
  hardcoded CHAR(32) the raw upgrade DDL produced, so it emits native UUID on
  PostgreSQL.

Index the hot filters [HIGH-04]
- Add composite Index("ix_job_status_date_created", "status", "date_created")
  for the worker poll, and index the foreign keys the worker and detail pages
  filter on: job.document_id, source.document_id, job_source.job_id,
  job_source.source_id, document.document_type_id, and the three
  document_person foreign keys.

Stop preloading by default [CRIT-02]
- Flip 16 relationships from lazy="selectin" to lazy="raise". The bidirectional
  selectin defaults meant loading one Job pulled a large connected subgraph.
- Three further relationships (ExecutionAttempt.job_source,
  ProcessingArtifact.execution_attempt, ProcessingArtifact.source) declared no
  lazy at all and defaulted to "select", which raises MissingGreenlet under
  async. These are now "raise" as well.
- Only 5 of 262 tests failed under the flip; the service layer already carried
  explicit eager loads. Fixes went into the service queries, never back into
  the models:
  - PeopleService._finalize_link refreshes document, person, and role_ref so
    the DocumentPerson write endpoints can still project them.
  - JobService.update_job_state loads job_sources -> source so the Job it
    returns still answers .error_detail and .filename.
  - Two tests that bypassed the service layer now load explicitly.
- Audited every UI relationship access against its feeding service method; all
  resolve to *_detail / list_*_detail variants with complete eager loads.

Tests
- Assert the composite and hot foreign key indexes exist in a fresh schema.
- Assert metadata.sorted_tables raises no SAWarning and orders source before
  execution_attempt.
- Assert preferred_execution_attempt_id is a Uuid that compiles to UUID on
  PostgreSQL and that its foreign key carries use_alter.
- Guard CRIT-02 from regression: no mapped relationship may declare a lazy
  strategy outside {raise, noload}.

The development database was rebuilt from metadata rather than upgraded; the
previous file is retained out of tree as the Phase 8 migration source.

Verified: 266 passed, 4 skipped; ruff check clean.

Co-authored-by: Copilot App <[email protected]>

src/transcription/db/__init__.py

@@ -1,5 +1,4 @@
 from .operations import create_all
-from .operations import upgrade_schema
 from .runtime import dispose_database_runtime
 from .runtime import initialize_database_runtime
 from .session import session_scope
@@ -11,5 +10,4 @@ __all__ = [
     "initialize_database_runtime",
     "session_scope",
     "transaction_scope",
-    "upgrade_schema",
 ]

src/transcription/db/models.py

@@ -14,8 +14,11 @@ from sqlalchemy import BigInteger
 from sqlalchemy import CheckConstraint
 from sqlalchemy import Column
 from sqlalchemy import Enum as SAEnum
+from sqlalchemy import ForeignKey
+from sqlalchemy import Index
 from sqlalchemy import LargeBinary
 from sqlalchemy import UniqueConstraint
+from sqlalchemy import Uuid
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm.exc import DetachedInstanceError
 from sqlalchemy.types import TypeDecorator
@@ -69,7 +72,7 @@ class DocumentType(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     documents: list["Document"] = Relationship(
-        back_populates="document_type_ref", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="document_type_ref", sa_relationship_kwargs={"lazy": "raise"}
     )
 
 
@@ -87,7 +90,7 @@ class PersonRole(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     document_people: list["DocumentPerson"] = Relationship(
-        back_populates="role_ref", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="role_ref", sa_relationship_kwargs={"lazy": "raise"}
     )
 
 
@@ -96,7 +99,7 @@ class Document(SQLModel, table=True):
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
     name: str
-    document_type_id: UUID | None = Field(default=None, foreign_key="document_type.id")
+    document_type_id: UUID | None = Field(default=None, foreign_key="document_type.id", index=True)
     document_date: date | None = None
     document_date_raw: str | None = None
     location_created: str | None = None
@@ -105,13 +108,13 @@ class Document(SQLModel, table=True):
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    jobs: list["Job"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"})
+    jobs: list["Job"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "raise"})
-    sources: list["Source"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"})
+    sources: list["Source"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "raise"})
     document_people: list["DocumentPerson"] = Relationship(
-        back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="document", sa_relationship_kwargs={"lazy": "raise"}
     )
     document_type_ref: Optional["DocumentType"] = Relationship(
-        back_populates="documents", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="documents", sa_relationship_kwargs={"lazy": "raise"}
     )
 
 
@@ -139,7 +142,7 @@ class Person(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     document_people: list["DocumentPerson"] = Relationship(
-        back_populates="person", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="person", sa_relationship_kwargs={"lazy": "raise"}
     )
 
 
@@ -149,30 +152,32 @@ class DocumentPerson(SQLModel, table=True):
     __tablename__ = "document_person"
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    document_id: UUID = Field(foreign_key="document.id")
+    document_id: UUID = Field(foreign_key="document.id", index=True)
-    person_id: UUID = Field(foreign_key="person.id")
+    person_id: UUID = Field(foreign_key="person.id", index=True)
-    role_id: UUID = Field(foreign_key="person_role.id")
+    role_id: UUID = Field(foreign_key="person_role.id", index=True)
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     __table_args__ = (UniqueConstraint("document_id", "person_id", name="uq_document_person"),)
 
     document: Optional["Document"] = Relationship(
-        back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="document_people", sa_relationship_kwargs={"lazy": "raise"}
     )
     person: Optional["Person"] = Relationship(
-        back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="document_people", sa_relationship_kwargs={"lazy": "raise"}
     )
     role_ref: Optional["PersonRole"] = Relationship(
-        back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}
+        back_populates="document_people", sa_relationship_kwargs={"lazy": "raise"}
     )
 
 
 class Job(SQLModel, table=True):
     """A transcription job tied to a single document."""
 
+    __table_args__ = (Index("ix_job_status_date_created", "status", "date_created"),)
+
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    document_id: UUID = Field(foreign_key="document.id")
+    document_id: UUID = Field(foreign_key="document.id", index=True)
     status: JobStatus = Field(
         default=JobStatus.QUEUED,
         sa_column=Column(
@@ -208,8 +213,8 @@ class Job(SQLModel, table=True):
     temperature: float | None = None
     top_p: float | None = None
 
-    document: Optional["Document"] = Relationship(back_populates="jobs", sa_relationship_kwargs={"lazy": "selectin"})
+    document: Optional["Document"] = Relationship(back_populates="jobs", sa_relationship_kwargs={"lazy": "raise"})
-    job_sources: list["JobSource"] = Relationship(back_populates="job", sa_relationship_kwargs={"lazy": "selectin"})
+    job_sources: list["JobSource"] = Relationship(back_populates="job", sa_relationship_kwargs={"lazy": "raise"})
 
     @property
     def filename(self) -> str:
@@ -249,7 +254,7 @@ class Source(SQLModel, table=True):
     """A document source image or PDF page."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    document_id: UUID = Field(foreign_key="document.id")
+    document_id: UUID = Field(foreign_key="document.id", index=True)
     page_number: int = Field(default=1, ge=1)
     upload_name: str
     filename: str
@@ -259,8 +264,18 @@ class Source(SQLModel, table=True):
     raw_transcription: str | None = None
     preferred_execution_attempt_id: UUID | None = Field(
         default=None,
-        foreign_key="execution_attempt.id",
+        sa_column=Column(
+            Uuid(),
+            # use_alter breaks the source / job_source / execution_attempt cycle so
+            # metadata.create_all can order table creation on every dialect.
+            ForeignKey(
+                "execution_attempt.id",
+                use_alter=True,
+                name="fk_source_preferred_execution_attempt_id",
+            ),
+            nullable=True,
             index=True,
+        ),
     )
     revised_text: str | None = None
     date_uploaded: datetime = Field(default_factory=lambda: datetime.now(UTC))
@@ -268,11 +283,11 @@ class Source(SQLModel, table=True):
 
     document: Optional["Document"] = Relationship(
         back_populates="sources",
-        sa_relationship_kwargs={"lazy": "selectin"},
+        sa_relationship_kwargs={"lazy": "raise"},
     )
     job_sources: list["JobSource"] = Relationship(
         back_populates="source",
-        sa_relationship_kwargs={"lazy": "selectin"},
+        sa_relationship_kwargs={"lazy": "raise"},
     )
     processing_artifacts: list["ProcessingArtifact"] = Relationship(
         back_populates="source",
@@ -310,8 +325,8 @@ class JobSource(SQLModel, table=True):
     __tablename__ = "job_source"
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    job_id: UUID = Field(foreign_key="job.id")
+    job_id: UUID = Field(foreign_key="job.id", index=True)
-    source_id: UUID = Field(foreign_key="source.id")
+    source_id: UUID = Field(foreign_key="source.id", index=True)
     status: JobSourceStatus = Field(
         default=JobSourceStatus.PENDING,
         sa_column=Column(
@@ -329,8 +344,8 @@ class JobSource(SQLModel, table=True):
     error_detail: str | None = None
     executed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"})
+    job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"})
-    source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"})
+    source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"})
     execution_attempts: list["ExecutionAttempt"] = Relationship(
         back_populates="job_source",
         sa_relationship_kwargs={"lazy": "noload", "order_by": "ExecutionAttempt.attempt_number"},
@@ -380,7 +395,9 @@ class ExecutionAttempt(SQLModel, table=True):
     duration_ms: int = Field(ge=0)
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    job_source: Optional["JobSource"] = Relationship(back_populates="execution_attempts")
+    job_source: Optional["JobSource"] = Relationship(
+        back_populates="execution_attempts", sa_relationship_kwargs={"lazy": "raise"}
+    )
     artifacts: list["ProcessingArtifact"] = Relationship(
         back_populates="execution_attempt", sa_relationship_kwargs={"lazy": "noload"}
     )
@@ -416,5 +433,9 @@ class ProcessingArtifact(SQLModel, table=True):
     )
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    execution_attempt: Optional["ExecutionAttempt"] = Relationship(back_populates="artifacts")
+    execution_attempt: Optional["ExecutionAttempt"] = Relationship(
-    source: Optional["Source"] = Relationship(back_populates="processing_artifacts")
+        back_populates="artifacts", sa_relationship_kwargs={"lazy": "raise"}
+    )
+    source: Optional["Source"] = Relationship(
+        back_populates="processing_artifacts", sa_relationship_kwargs={"lazy": "raise"}
+    )

src/transcription/db/operations.py

@@ -2,9 +2,6 @@ from __future__ import annotations
 
 import logging
 
-from sqlalchemy import inspect
-from sqlalchemy import text
-from sqlalchemy.ext.asyncio import AsyncConnection
 from sqlalchemy.ext.asyncio import AsyncEngine
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlmodel import SQLModel
@@ -13,8 +10,6 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 
 from .engine import resolve_engine
 from .models import DocumentType
-from .models import Job
-from .models import JobStatus
 from .models import PersonRole
 from .registries import BUILT_IN_DOCUMENT_TYPES
 from .registries import BUILT_IN_PERSON_ROLES
@@ -30,85 +25,10 @@ async def create_all(*, engine: AsyncEngine | None = None) -> None:
     active_engine = engine or resolve_engine()
     async with active_engine.begin() as connection:
         await connection.run_sync(SQLModel.metadata.create_all)
-        await _upgrade_person_family_search_id(connection)
-        await _upgrade_v42_evidence_tables(connection)
-        await _upgrade_v45_selection_columns(connection)
     await seed_registry_defaults(engine=active_engine)
     logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url)
 
 
-async def upgrade_schema(*, engine: AsyncEngine | None = None) -> None:
-    """Apply non-destructive additive upgrades to an existing schema."""
-    active_engine = engine or resolve_engine()
-    async with active_engine.begin() as connection:
-        await _upgrade_person_family_search_id(connection)
-        await _upgrade_v42_evidence_tables(connection)
-        await _upgrade_v45_selection_columns(connection)
-
-
-async def _upgrade_v42_evidence_tables(connection: AsyncConnection) -> None:
-    """Create the additive V4.2 evidence tables without rewriting historical rows."""
-
-    def create_tables(sync_connection) -> None:
-        SQLModel.metadata.tables["execution_attempt"].create(sync_connection, checkfirst=True)
-        SQLModel.metadata.tables["processing_artifact"].create(sync_connection, checkfirst=True)
-
-    await connection.run_sync(create_tables)
-
-
-async def _upgrade_v45_selection_columns(connection: AsyncConnection) -> None:
-    """Add V4.5 purpose and preferred-attempt provenance columns."""
-
-    def inspect_columns(sync_connection) -> tuple[set[str], set[str]]:
-        database = inspect(sync_connection)
-        tables = set(database.get_table_names())
-        job_columns = {column["name"] for column in database.get_columns("job")} if "job" in tables else set()
-        source_columns = (
-            {column["name"] for column in database.get_columns("source")} if "source" in tables else set()
-        )
-        return job_columns, source_columns
-
-    job_columns, source_columns = await connection.run_sync(inspect_columns)
-    if job_columns and "purpose" not in job_columns:
-        await connection.execute(
-            text("ALTER TABLE job ADD COLUMN purpose VARCHAR NOT NULL DEFAULT 'transcription'")
-        )
-    if source_columns and "preferred_execution_attempt_id" not in source_columns:
-        await connection.execute(text("ALTER TABLE source ADD COLUMN preferred_execution_attempt_id CHAR(32)"))
-        await connection.execute(
-            text(
-                "CREATE INDEX IF NOT EXISTS ix_source_preferred_execution_attempt_id "
-                "ON source (preferred_execution_attempt_id)"
-            )
-        )
-
-
-async def _upgrade_person_family_search_id(connection: AsyncConnection) -> None:
-    """Add the nullable V4.1 FamilySearch field to an existing database."""
-
-    def inspect_person(sync_connection) -> tuple[bool, bool]:
-        database = inspect(sync_connection)
-        if "person" not in database.get_table_names():
-            return False, False
-        columns = {column["name"] for column in database.get_columns("person")}
-        indexes = database.get_indexes("person")
-        constraints = database.get_unique_constraints("person")
-        has_unique_id = any(entry.get("column_names") == ["family_search_id"] for entry in [*indexes, *constraints])
-        return "family_search_id" in columns, has_unique_id
-
-    has_column, has_unique_id = await connection.run_sync(inspect_person)
-    if not has_column and not await connection.run_sync(
-        lambda sync_connection: "person" in inspect(sync_connection).get_table_names()
-    ):
-        return
-    if not has_column:
-        await connection.execute(text("ALTER TABLE person ADD COLUMN family_search_id VARCHAR"))
-    if not has_unique_id:
-        await connection.execute(
-            text("CREATE UNIQUE INDEX IF NOT EXISTS ix_person_family_search_id ON person (family_search_id)")
-        )
-
-
 async def seed_registry_defaults(*, engine: AsyncEngine | None = None) -> None:
     """Seed default registry rows for role and document type taxonomies."""
     active_engine = engine or resolve_engine()
@@ -138,14 +58,3 @@ async def seed_registry_defaults(*, engine: AsyncEngine | None = None) -> None:
                 )
 
         await session.commit()
-
-
-async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
-    """Get the next queued job, if any."""
-    result = await session.exec(
-        select(Job)
-        .where(Job.status == JobStatus.QUEUED)
-        .order_by(Job.date_created)  # pyright: ignore[reportArgumentType]
-        .limit(1)
-    )  # fmt: skip
-    return result.first()

src/transcription/services/jobs.py

@@ -113,11 +113,9 @@ class JobService(ServiceBase):
     async def list_jobs(
         self,
         *,
-        load_docs: bool = False,
         session: AsyncSession | None = None,
     ) -> Sequence[Job]:
         """List all jobs in the database with eagerly loaded documents."""
-        _ = load_docs
         async with self._session_scope(session) as _session:
             query = select(Job).options(
                 selectinload(Job.document),  # pyright: ignore[reportArgumentType]
@@ -153,7 +151,10 @@ class JobService(ServiceBase):
         async with self._session_scope(session) as _session:
             query = (
                 select(Job)
-                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .options(
+                    selectinload(Job.document),  # pyright: ignore[reportArgumentType]
+                    selectinload(Job.job_sources).selectinload(JobSource.source),  # pyright: ignore[reportArgumentType]
+                )
                 .where(Job.id == job_id)
                 .execution_options(populate_existing=True)
             )

src/transcription/services/people.py

@@ -570,6 +570,8 @@ class PeopleService(ServiceBase):
                 category=ErrorCategory.CONFLICT,
                 suggestion="Edit the existing relationship instead of adding another one.",
             ) from exc
+        # Relationships load explicitly; the models declare lazy="raise".
+        await session.refresh(link, attribute_names=["document", "person", "role_ref"])
         return link
 
     async def _require_document(self, *, session: AsyncSession, document_id: UUID) -> None:

tests/test_db.py

@@ -1,8 +1,12 @@
 """Tests for the database runtime and V2 schema bootstrap behavior."""
 
+import warnings
+
 import pytest
+import sqlalchemy as sa
 from sqlalchemy import inspect
-from sqlalchemy import text
+from sqlalchemy.dialects import postgresql
+from sqlmodel import SQLModel
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -12,9 +16,9 @@ from transcription.db import create_all
 from transcription.db import dispose_database_runtime
 from transcription.db import initialize_database_runtime
 from transcription.db import session_scope
-from transcription.db import upgrade_schema
 from transcription.db.models import DocumentType
 from transcription.db.models import PersonRole
+from transcription.db.models import Source
 
 
 @pytest.mark.asyncio
@@ -106,92 +110,61 @@ async def test_create_all_seeds_default_registry_rows(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_create_all_upgrades_existing_person_table_for_family_search(tmp_path):
+async def test_create_all_declares_hot_path_indexes(tmp_path):
+    """Worker and detail-page filters must be index-backed in a freshly created schema."""
     settings = Settings(
         openrouter_api_key="test-key",
-        database=SqliteSettings(path=str(tmp_path / "upgrade.db")),
+        database=SqliteSettings(path=str(tmp_path / "indexes.db")),
         environment="test",
     )
     runtime = initialize_database_runtime(settings=settings)
 
     try:
-        async with runtime.engine.begin() as connection:
-            await connection.execute(
-                text("CREATE TABLE person (id CHAR(32) PRIMARY KEY NOT NULL, full_name VARCHAR NOT NULL)")
-            )
-
         await create_all(engine=runtime.engine)
         async with runtime.engine.connect() as connection:
-            columns, indexes = await connection.run_sync(
-                lambda sync_connection: (
-                    {column["name"] for column in inspect(sync_connection).get_columns("person")},
-                    inspect(sync_connection).get_indexes("person"),
-                )
-            )
 
-        assert "family_search_id" in columns
+            def collect(sync_connection) -> dict[str, list[list[str]]]:
-        assert any(index["column_names"] == ["family_search_id"] and index["unique"] for index in indexes)
+                database = inspect(sync_connection)
+                return {
+                    table: [index["column_names"] for index in database.get_indexes(table)]
+                    for table in ("job", "source", "job_source", "document", "document_person")
+                }
+
+            indexes = await connection.run_sync(collect)
+
+        assert ["status", "date_created"] in indexes["job"]
+        assert ["document_id"] in indexes["job"]
+        assert ["document_id"] in indexes["source"]
+        assert ["preferred_execution_attempt_id"] in indexes["source"]
+        assert ["job_id"] in indexes["job_source"]
+        assert ["source_id"] in indexes["job_source"]
+        assert ["document_type_id"] in indexes["document"]
+        for column in ("document_id", "person_id", "role_id"):
+            assert [column] in indexes["document_person"]
     finally:
         await dispose_database_runtime()
 
 
-@pytest.mark.asyncio
+def test_metadata_has_no_unresolvable_table_cycle():
-async def test_v42_upgrade_adds_evidence_tables_without_rewriting_legacy_snapshot(tmp_path):
+    """create_all must be able to order every table, including on PostgreSQL."""
-    settings = Settings(
+    with warnings.catch_warnings():
-        openrouter_api_key="test-key",
+        warnings.simplefilter("error", sa.exc.SAWarning)
-        database=SqliteSettings(path=str(tmp_path / "v42-upgrade.db")),
+        ordered = [table.name for table in SQLModel.metadata.sorted_tables]
-        environment="test",
-    )
-    runtime = initialize_database_runtime(settings=settings)
 
-    try:
+    assert ordered.index("source") < ordered.index("execution_attempt")
-        async with runtime.engine.begin() as connection:
-            await connection.execute(text("CREATE TABLE job (id CHAR(32) PRIMARY KEY NOT NULL)"))
-            await connection.execute(text("CREATE TABLE source (id CHAR(32) PRIMARY KEY NOT NULL)"))
-            await connection.execute(
-                text(
-                    "CREATE TABLE job_source ("
-                    "id CHAR(32) PRIMARY KEY NOT NULL, "
-                    "job_id CHAR(32) NOT NULL, "
-                    "source_id CHAR(32) NOT NULL, "
-                    "raw_api_response JSON"
-                    ")"
-                )
-            )
-            await connection.execute(text("INSERT INTO job (id) VALUES ('job-1')"))
-            await connection.execute(text("INSERT INTO source (id) VALUES ('source-1')"))
-            await connection.execute(
-                text(
-                    "INSERT INTO job_source (id, job_id, source_id, raw_api_response) "
-                    "VALUES ('link-1', 'job-1', 'source-1', :snapshot)"
-                ),
-                {"snapshot": '{"legacy":true}'},
-            )
 
-        await upgrade_schema(engine=runtime.engine)
-        await upgrade_schema(engine=runtime.engine)
-        async with runtime.engine.connect() as connection:
-            table_names = set(await connection.run_sync(lambda c: inspect(c).get_table_names()))
-            job_columns = set(
-                await connection.run_sync(
-                    lambda c: tuple(column["name"] for column in inspect(c).get_columns("job"))
-                )
-            )
-            source_columns = set(
-                await connection.run_sync(
-                    lambda c: tuple(column["name"] for column in inspect(c).get_columns("source"))
-                )
-            )
-            legacy_snapshot = (
-                await connection.execute(text("SELECT raw_api_response FROM job_source WHERE id = 'link-1'"))
-            ).scalar_one()
 
-        assert {"execution_attempt", "processing_artifact"}.issubset(table_names)
+def test_preferred_execution_attempt_id_column_matches_model_declaration():
-        assert "purpose" in job_columns
+    """The self-referential provenance column is a real UUID, not an opaque CHAR(32)."""
-        assert "preferred_execution_attempt_id" in source_columns
+    column = SQLModel.metadata.tables["source"].c["preferred_execution_attempt_id"]
-        assert "legacy" in legacy_snapshot
+
-    finally:
+    assert isinstance(column.type, sa.Uuid)
-        await dispose_database_runtime()
+    assert column.type.compile(dialect=postgresql.dialect()) == "UUID"
+    assert Source.model_fields["preferred_execution_attempt_id"].annotation is not None
+
+    foreign_key = next(iter(column.foreign_keys))
+    assert foreign_key.use_alter is True
+    assert foreign_key.column is SQLModel.metadata.tables["execution_attempt"].c["id"]
 
 
 def test_bootstrap_policy_production_defaults_false():

tests/test_models.py

@@ -246,7 +246,7 @@ class TestRelationships:
         session.add(link)
         session.commit()
 
-        session.refresh(document)
+        session.refresh(document, attribute_names=["jobs", "sources", "document_people"])
         assert len(document.jobs) == 1
         assert len(document.sources) == 1
         assert len(document.document_people) == 1
@@ -266,3 +266,16 @@ class TestRegistryModels:
         session.add(duplicate)
         with pytest.raises(IntegrityError):
             session.commit()
+
+
+def test_no_relationship_declares_an_implicit_eager_load():
+    """CRIT-02 guard: eager loading is a per-query decision, never a model default."""
+    from sqlmodel import SQLModel
+
+    offenders = {
+        f"{mapper.class_.__name__}.{relationship.key}": relationship.lazy
+        for mapper in SQLModel._sa_registry.mappers
+        for relationship in mapper.relationships
+        if relationship.lazy not in {"raise", "noload"}
+    }
+    assert offenders == {}, f"Relationships must not preload by default: {offenders}"

tests/ui/test_sources_page.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 import pytest
+from sqlalchemy.orm import selectinload
 from sqlmodel import select
 
 from transcription.db import session_scope
@@ -48,7 +49,16 @@ class TestSourceModelProperties:
         async with session_scope() as session:
             job = await session.get(Job, job_id)
             assert job is not None
-            source = (await session.exec(select(Source).where(Source.document_id == job.document_id))).first()
+            source = (
+                await session.exec(
+                    select(Source)
+                    .options(
+                        selectinload(Source.document),
+                        selectinload(Source.job_sources),
+                    )
+                    .where(Source.document_id == job.document_id)
+                )
+            ).first()
             assert source is not None
 
             # Validate computed properties