diff --git a/docs/requirements.md b/docs/requirements.md
index f616583..62dcf09 100644
--- a/docs/requirements.md
+++ b/docs/requirements.md
@@ -27,7 +27,7 @@ This page captures a SysML v1.6-style requirements baseline for the production s
 | REQ-10 | Design Constraint | Keep schema bootstrap explicit and opt-in; normal startup does not mutate production schema. | high | inspection |
 | REQ-11 | Design Constraint | Use service-backed persistence for core document and job data. | medium | inspection |
 | REQ-12 | Design Constraint | Store transcription prompts as individual Markdown artifacts for iterative refinement. | medium | inspection |
-| REQ-13 | Functional | Allow users to revise transcription text from either original text or a previous revision. | low | test |
+| REQ-13 | Functional | Allow users to create one optional revision of transcription text derived from the original job transcription. | low | test |
 
 ### Requirement Relationships
 
diff --git a/docs/schema.md b/docs/schema.md
index 2a5614b..d7c630e 100644
--- a/docs/schema.md
+++ b/docs/schema.md
@@ -58,6 +58,7 @@ erDiagram
 * A source can belong to only one job (which contains the original transcription). A source can only belong to one document. A source may have one optional transcription revision.
 * A job can process one or more sources. A job can belong to only one document.
 * A revision can belong to only one source. A source may have one optional revision.
+* 1:1 optionality is enforced by uniqueness on `revision.source_id` (no revision history chain).
 * `Job.text` stores the original immutable provider transcription.
 * Revision rows are optional user-authored edits and are derived from the original transcription. Unlike jobs, revision rows can be updated.
 
diff --git a/docs/ver1/ver1.md b/docs/ver1/ver1.md
index ad27827..3d16674 100644
--- a/docs/ver1/ver1.md
+++ b/docs/ver1/ver1.md
@@ -16,7 +16,7 @@ V1 is complete when all of the following are true:
 
 1. **Functional complete**
    - Upload, queue, processing, status display, and transcription result inspection work end-to-end.
-   - Optional revision workflow is implemented (create/list/view).
+   - Optional revision workflow is implemented (create/view/update single revision).
 2. **Data-model complete**
    - Runtime behavior, persistence, and tests all align to `Document` / `Source` / `Job` / `Revision`.
 3. **Operational complete**
@@ -83,7 +83,7 @@ V1 is complete when all of the following are true:
 ### Tasks
 1. Update job detail and related UI components:
    - Display original immutable transcription from `Job.text`.
-   - Display optional revision sourced from `Source.revisions`.
+   - Display optional revision sourced from `Source.revision` (0 or 1).
 2. Align date fields with new schema naming.
 3. Preserve clear user messaging when no revisions exist.
 
@@ -125,7 +125,7 @@ V1 is complete when all of the following are true:
    - `Document`, `Source`, `Job`, `Revision` relationships and invariants.
 2. Rewrite service/integration tests:
    - Worker success/failure paths using `Job.text` / `Job.error_detail`.
-   - Optional revision creation and lineage behavior.
+   - Optional single-revision creation/update behavior.
 3. Update UI tests for new job-detail/revision rendering behavior.
 4. Re-enable strict CI quality gates (lint, type, tests).
 
diff --git a/src/transcription/db/operations.py b/src/transcription/db/operations.py
index 00919e1..3a5dcb8 100644
--- a/src/transcription/db/operations.py
+++ b/src/transcription/db/operations.py
@@ -22,7 +22,7 @@ async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
     result = await session.exec(
         select(Job)
         .where(Job.status == JobStatus.QUEUED)
-        .order_by(Job.created_at)  # pyright: ignore[reportArgumentType]
+        .order_by(Job.date_created)  # pyright: ignore[reportArgumentType]
         .limit(1)
     )  # fmt: skip
     return result.first()
@@ -58,8 +58,14 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
             connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
             logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
 
-    if "transcript" in table_names:
-        transcript_columns = {column["name"] for column in inspector.get_columns("transcript")}
-        if "model" not in transcript_columns:
-            connection.execute(text("ALTER TABLE transcript ADD COLUMN model VARCHAR NOT NULL DEFAULT 'unknown'"))
-            logger.warning("Applied SQLite compatibility schema patch table=transcript column=model default=unknown")
+    if "revision" in table_names:
+        revision_columns = {column["name"] for column in inspector.get_columns("revision")}
+        if "source_id" in revision_columns:
+            has_unique_source = False
+            for index in inspector.get_indexes("revision"):
+                if index.get("unique") and index.get("column_names") == ["source_id"]:
+                    has_unique_source = True
+                    break
+            if not has_unique_source:
+                connection.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ux_revision_source_id ON revision(source_id)"))
+                logger.warning("Applied SQLite compatibility schema patch table=revision unique_index=ux_revision_source_id")
diff --git a/src/transcription/models.py b/src/transcription/models.py
index ca30bbf..89f96d8 100644
--- a/src/transcription/models.py
+++ b/src/transcription/models.py
@@ -1,7 +1,9 @@
 """SQLModel domain models for the transcription system.
 
-Three models capture the MVP lifecycle:
-    Document  ->  one-to-many  ->  Source -> one-to-many  ->  Job  ->  one-to-many  ->  Transcript
+Core V1 lifecycle:
+    Document  ->  one-to-many  ->  Source
+    Document  ->  one-to-many  ->  Job
+    Source    ->  one-to-one?  ->  Revision (optional)
 """
 
 from datetime import UTC
@@ -36,7 +38,7 @@ class Document(SQLModel, table=True):
 
 
 class Source(SQLModel, table=True):
-    """A document source (image or pdf)."""
+    """A document source (image or PDF)."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
     document_id: UUID = Field(foreign_key="document.id")
@@ -52,7 +54,10 @@ class Source(SQLModel, table=True):
     # Relationships
     document: Optional["Document"] = Relationship(back_populates="sources")
     job: Optional["Job"] = Relationship(back_populates="sources")
-    revisions: list["Revision"] = Relationship(back_populates="source")
+    revision: "Revision | None" = Relationship(
+        back_populates="source",
+        sa_relationship_kwargs={"uselist": False},
+    )
 
 
 class Job(SQLModel, table=True):
@@ -64,11 +69,11 @@ class Job(SQLModel, table=True):
     retry_count: int = Field(default=0, ge=0)
     date_created: datetime = Field(default_factory=lambda: datetime.now(UTC))
     date_updated: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    provider: str
+    provider: str | None = None
     """Name of the transcription provider used to generate this transcript."""
-    model: str
+    model: str | None = None
     """Model identifier used to generate this transcript."""
-    prompt_name: str
+    prompt_name: str | None = None
     """Name of the prompt used to generate this transcript."""
     text: str | None = None
     """The transcribed text. This may be None if the job failed or is still in progress."""
@@ -81,8 +86,10 @@ class Job(SQLModel, table=True):
 
     @property
     def filename(self) -> str:
-        """Return the filename of the associated document."""
-        return self.source.filename if self.source else "unknown"
+        """Return the filename of the associated source, when available."""
+        if not self.sources:
+            return "unknown"
+        return self.sources[0].filename
 
 
 class Revision(SQLModel, table=True):
@@ -97,7 +104,7 @@ class Revision(SQLModel, table=True):
     """The revised text."""
     date_created: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    # __table_args__ = (UniqueConstraint("job_id", "revision", name="uq_transcript_job_revision"),)
+    __table_args__ = (UniqueConstraint("source_id", name="uq_revision_source_id"),)
 
     # Relationships
-    source: Optional["Source"] = Relationship(back_populates="revisions")
+    source: Optional["Source"] = Relationship(back_populates="revision")
diff --git a/src/transcription/providers/base.py b/src/transcription/providers/base.py
index 25a017e..9056f45 100644
--- a/src/transcription/providers/base.py
+++ b/src/transcription/providers/base.py
@@ -2,9 +2,6 @@
 
 from dataclasses import dataclass
 from typing import Protocol
-from uuid import UUID
-
-from ..models import Transcript
 
 
 class ProviderError(RuntimeError):
@@ -28,17 +25,6 @@ class TranscriptionResult:
     prompt_name: str
     model: str
 
-    def to_transcript(self, job_id: UUID, *, revision: int = 0) -> Transcript:
-        """Convert a TranscriptionResult to a Transcript model instance."""
-        return Transcript(
-            job_id=job_id,
-            revision=revision,
-            provider=self.provider,
-            prompt_name=self.prompt_name,
-            model=self.model,
-            text=self.text,
-        )
-
 
 class TranscriptionProvider(Protocol):
     """Contract every transcription provider adapter must satisfy."""
diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py
index 03a66f0..d67a94d 100644
--- a/src/transcription/services/documents.py
+++ b/src/transcription/services/documents.py
@@ -21,8 +21,8 @@ class DocumentError(AppError):
     """Raised when document operations fail."""
 
 
-class MissingImageError(DocumentError):
-    """Raised when a required image is missing."""
+class MissingSourceError(DocumentError):
+    """Raised when a document has no associated sources."""
 
 
 class UploadError(DocumentError):
@@ -30,7 +30,7 @@ class UploadError(DocumentError):
 
 
 class DocumentAlreadyExistsError(DocumentError):
-    """Raised when a document with the same filename already exists in the database."""
+    """Raised when a document with the same name already exists in the database."""
 
 
 @dataclass(frozen=True)
@@ -72,13 +72,16 @@ class DocumentService(ServiceBase):
     async def read_document(self, document_id: UUID, *, session: AsyncSession | None = None) -> Document:
         """Read an existing document from the database.
 
-        The selectinload option is used to eagerly load related jobs for the document.
+        The selectinload option is used to eagerly load related jobs and sources.
         """
         async with self._session_scope(session) as _session:
             document = await _session.get(
                 Document,
                 document_id,
-                options=(selectinload(Document.jobs),),  # pyright: ignore[reportArgumentType]
+                options=(
+                    selectinload(Document.jobs),  # pyright: ignore[reportArgumentType]
+                    selectinload(Document.sources),  # pyright: ignore[reportArgumentType]
+                ),
             )
         if document is None:
             raise DocumentError(
@@ -86,11 +89,11 @@ class DocumentService(ServiceBase):
                 category=ErrorCategory.NOT_FOUND,
                 suggestion="Re-upload the source document and retry.",
             )
-        elif not Path(document.file_path).exists():
-            raise MissingImageError(
-                f"Document with id {document_id} is missing its image file in {self.settings.upload_dir:!s}",
+        elif not document.sources:
+            raise MissingSourceError(
+                f"Document with id {document_id} has no associated source records",
                 category=ErrorCategory.NOT_FOUND,
-                suggestion="Re-upload the source document and retry.",
+                suggestion="Upload at least one source for this document and retry.",
             )
         return document
 
@@ -110,13 +113,13 @@ class DocumentService(ServiceBase):
     # Query Operations
 
     async def query_documents(
-        self, *, filename: str | None = None, session: AsyncSession | None = None
+        self, *, name: str | None = None, session: AsyncSession | None = None
     ) -> Sequence[Document]:
         """Query documents from the database based on provided filters."""
         async with self._session_scope(session) as _session:
             query = select(Document)
-            if filename is not None:
-                query = query.where(Document.filename == filename)
+            if name is not None:
+                query = query.where(Document.name == name)
             result = await _session.exec(query)
             return result.all()
 
diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py
index 8d94321..66a8a56 100644
--- a/src/transcription/services/jobs.py
+++ b/src/transcription/services/jobs.py
@@ -9,6 +9,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..models import Job
 from ..models import JobStatus
+from ..models import Source
 from .base import ServiceBase
 
 
@@ -37,7 +38,7 @@ class JobService(ServiceBase):
                 select(Job)
                 .options(
                     selectinload(Job.document),  # pyright: ignore[reportArgumentType]
-                    selectinload(Job.transcripts),  # pyright: ignore[reportArgumentType]
+                    selectinload(Job.sources).selectinload(Source.revision),  # pyright: ignore[reportArgumentType]
                 )
                 .where(Job.id == job_id)
                 .execution_options(populate_existing=True)
@@ -71,11 +72,14 @@ class JobService(ServiceBase):
     ) -> Sequence[Job]:
         """Query jobs from the database based on provided filters."""
         async with self._session_scope(session) as _session:
-            query = select(Job).options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+            query = select(Job).options(
+                selectinload(Job.document),  # pyright: ignore[reportArgumentType]
+                selectinload(Job.sources),  # pyright: ignore[reportArgumentType]
+            )
             if status is not None:
                 query = query.where(Job.status == status)
             if filename is not None:
-                query = query.where(Job.document.filename == filename)
+                query = query.where(Job.sources.any(Source.filename == filename))
             result = await _session.exec(query)
             return result.all()
 
@@ -88,7 +92,10 @@ class JobService(ServiceBase):
         """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]
+            query = select(Job).options(
+                selectinload(Job.document),  # pyright: ignore[reportArgumentType]
+                selectinload(Job.sources),  # pyright: ignore[reportArgumentType]
+            )
             result = await _session.exec(query)
             return result.all()
 
@@ -129,7 +136,7 @@ class JobService(ServiceBase):
             job.status = status
             if retry_count_increment:
                 job.retry_count += retry_count_increment
-            job.updated_at = datetime.now(UTC)
+            job.date_updated = datetime.now(UTC)
             await self._finalize(session=_session, caller_session=session, refresh=(job,))
             return job
 
@@ -144,6 +151,6 @@ class JobService(ServiceBase):
                 select(Job)
                 .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
                 .where(Job.status == JobStatus.QUEUED)
-                .order_by(Job.created_at)  # pyright: ignore[reportArgumentType]
+                .order_by(Job.date_created)  # pyright: ignore[reportArgumentType]
             )
             return (await _session.exec(query)).first()
diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py
index aeda0e3..95eb293 100644
--- a/src/transcription/services/store.py
+++ b/src/transcription/services/store.py
@@ -13,6 +13,7 @@ from transcription.errors import ErrorCategory
 
 from ..models import Document
 from ..models import Job
+from ..models import Source
 from .documents import UploadJobResult
 
 logger = logging.getLogger(__name__)
@@ -69,14 +70,24 @@ async def _create_upload_records(
     stored_path: Path,
 ) -> tuple[Document, Job]:
     document = Document(
-        filename=Path(original_filename).name,
-        file_path=str(stored_path),
+        name=Path(original_filename).name,
     )
     session.add(document)
     await session.flush()
 
     job = Job(document_id=document.id)
     session.add(job)
+    await session.flush()
+
+    source = Source(
+        document_id=document.id,
+        job_id=job.id,
+        upload_name=Path(original_filename).name,
+        filename=stored_path.name,
+        file_path=str(stored_path),
+    )
+    session.add(source)
+
     await session.commit()
     await session.refresh(document)
     await session.refresh(job)
diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py
index 3225080..9475c8a 100644
--- a/src/transcription/services/transcription.py
+++ b/src/transcription/services/transcription.py
@@ -6,10 +6,11 @@ import logging
 import mimetypes
 from collections.abc import Sequence
 from contextlib import contextmanager
+from datetime import UTC
+from datetime import datetime
 from pathlib import Path
 from uuid import UUID
 
-from sqlalchemy import func
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlalchemy.orm import selectinload
 from sqlmodel import select
@@ -19,7 +20,9 @@ from transcription.config import Settings
 from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
-from transcription.models import Transcript
+from transcription.models import Job
+from transcription.models import Revision
+from transcription.models import Source
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
 from transcription.providers import ProviderResponseError
@@ -44,13 +47,11 @@ class TranscriptionError(AppError):
 
 
 class TranscriptionNotFoundError(TranscriptionError):
-    """Raised when a transcription is not found in the database."""
+    """Raised when a transcription-related resource is not found."""
 
 
 class TranscriptionService(ServiceBase):
-    """Service class for managing transcription operations.
-
-    This is the top-level service that composes functionality from the other services."""
+    """Service class for job transcription output and optional source revisions."""
 
     provider: TranscriptionProvider
 
@@ -58,43 +59,52 @@ class TranscriptionService(ServiceBase):
         super().__init__(session_factory=session_factory)
         self.provider = get_transcription_provider(settings=self.settings)
 
-    async def create_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> Transcript:
-        """Create a new transcript in the database."""
+    async def create_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> Revision:
+        """Create a new revision in the database."""
         async with self._session_scope(session) as _session:
-            _session.add(transcript)
-            await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
-            return transcript
+            _session.add(revision)
+            await self._finalize(session=_session, caller_session=session, refresh=(revision,))
+            return revision
 
-    async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Transcript:
-        """Read an existing transcript from the database."""
+    async def read_revision(self, revision_id: UUID, *, session: AsyncSession | None = None) -> Revision:
+        """Read an existing revision from the database."""
         async with self._session_scope(session) as _session:
-            transcript = await _session.get(
-                Transcript,
-                transcript_id,
-                # Makes the full Job model object available in the return Transcript object
-                options=(selectinload(Transcript.job),),  # pyright: ignore[reportArgumentType]
+            revision = await _session.get(
+                Revision,
+                revision_id,
+                options=(selectinload(Revision.source),),  # pyright: ignore[reportArgumentType]
             )
-        if transcript is None:
+        if revision is None:
             raise TranscriptionNotFoundError(
-                f"Transcript with id {transcript_id} not found",
+                f"Revision with id {revision_id} not found",
                 category=ErrorCategory.NOT_FOUND,
-                suggestion="Verify the transcript id and retry.",
+                suggestion="Verify the revision id and retry.",
             )
-        return transcript
+        return revision
 
-    async def update_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> Transcript:
-        """Update an existing transcript in the database."""
+    async def update_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> Revision:
+        """Update an existing revision in the database."""
         async with self._session_scope(session) as _session:
-            merged = await _session.merge(transcript)
+            merged = await _session.merge(revision)
             await self._finalize(session=_session, caller_session=session, refresh=(merged,))
             return merged
 
-    async def delete_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> None:
-        """Delete a transcript from the database."""
+    async def delete_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> None:
+        """Delete a revision from the database."""
         async with self._session_scope(session) as _session:
-            await _session.delete(transcript)
+            await _session.delete(revision)
             await self._finalize(session=_session, caller_session=session)
 
+    # Temporary compatibility methods for callers still using transcript naming.
+
+    async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Revision:
+        """Backward-compatible alias for read_revision."""
+        return await self.read_revision(transcript_id, session=session)
+
+    async def delete_transcript(self, transcript: Revision, *, session: AsyncSession | None = None) -> None:
+        """Backward-compatible alias for delete_revision."""
+        await self.delete_revision(transcript, session=session)
+
     async def transcribe_document(
         self,
         image_path: str | Path,
@@ -102,7 +112,7 @@ class TranscriptionService(ServiceBase):
         *,
         prompt_name: str = DEFAULT_PROMPT_FILE,
         session: AsyncSession | None = None,
-    ):
+    ) -> None:
         """Transcribe a local image using the configured prompt and provider."""
         result = await transcribe_document_image(
             image_path=image_path,
@@ -110,16 +120,17 @@ class TranscriptionService(ServiceBase):
             settings=self.settings,
             provider=self.provider,
         )
-        await self.create_transcript_for_job(
+        await self.update_job_transcription(
             job_id=job_id,
             text=result.text,
+            error_detail=None,
             provider=result.provider,
             model=result.model,
             prompt_name=result.prompt_name,
             session=session,
         )
 
-    async def create_transcript_for_job(
+    async def update_job_transcription(
         self,
         *,
         job_id: UUID,
@@ -129,61 +140,88 @@ class TranscriptionService(ServiceBase):
         model: str | None = None,
         prompt_name: str = DEFAULT_PROMPT_FILE,
         session: AsyncSession | None = None,
-    ) -> Transcript:
-        """Create a new transcript revision for a job id."""
+    ) -> Job:
+        """Persist original transcription output fields on a job."""
         async with self._session_scope(session) as _session:
-            rev_query = select(func.max(Transcript.revision)).where(Transcript.job_id == job_id)
-            rev_result = await _session.exec(rev_query)
-            max_revision = -1 if (rev := rev_result.one_or_none()) is None else rev
-            next_revision = max_revision + 1
+            job = await _session.get(Job, job_id)
+            if job is None:
+                raise TranscriptionNotFoundError(
+                    f"Job with id {job_id} not found",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Verify the job id and retry.",
+                )
 
-            transcript = Transcript(
-                job_id=job_id,
-                revision=next_revision,
-                provider=provider or self.settings.provider.value,
-                model=model or _resolve_transcript_model(provider=self.provider, settings=self.settings),
-                prompt_name=prompt_name,
-                text=text,
-                error_detail=error_detail,
-            )
-            _session.add(transcript)
-            await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
-            return transcript
+            job.text = text
+            job.error_detail = error_detail
+            job.provider = provider or job.provider or self.settings.provider.value
+            job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
+            job.prompt_name = prompt_name or job.prompt_name or DEFAULT_PROMPT_FILE
+            job.date_updated = datetime.now(UTC)
 
-    async def read_latest_transcript_by_job(
+            await self._finalize(session=_session, caller_session=session, refresh=(job,))
+            return job
+
+    async def upsert_revision_for_source(
         self,
-        job_id: UUID,
+        *,
+        source_id: UUID,
+        text: str,
+        session: AsyncSession | None = None,
+    ) -> Revision:
+        """Create or replace the single optional revision for a source."""
+        async with self._session_scope(session) as _session:
+            source = await _session.get(Source, source_id)
+            if source is None:
+                raise TranscriptionNotFoundError(
+                    f"Source with id {source_id} not found",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Verify the source id and retry.",
+                )
+
+            query = select(Revision).where(Revision.source_id == source_id)
+            existing = (await _session.exec(query)).one_or_none()
+
+            if existing is None:
+                revision = Revision(source_id=source_id, text=text)
+                _session.add(revision)
+                await self._finalize(session=_session, caller_session=session, refresh=(revision,))
+                return revision
+
+            existing.text = text
+            merged = await _session.merge(existing)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def read_revision_by_source(
+        self,
+        source_id: UUID,
         *,
         session: AsyncSession | None = None,
-    ) -> Transcript | None:
-        """Read the latest transcript revision for a job id."""
+    ) -> Revision | None:
+        """Read the single optional revision for a source."""
         async with self._session_scope(session) as _session:
-            query = _transcript_job_query(job_id=job_id).limit(1)
+            query = select(Revision).where(Revision.source_id == source_id)
             result = await _session.exec(query)
             return result.one_or_none()
 
-    async def list_transcripts_by_job(
+    async def list_revisions_by_job(
         self,
         job_id: UUID,
         *,
         session: AsyncSession | None = None,
-    ) -> Sequence[Transcript]:
-        """List transcript revisions for a job id in ascending revision order."""
+    ) -> Sequence[Revision]:
+        """List revisions connected to all sources for a job."""
         async with self._session_scope(session) as _session:
-            query = _transcript_job_query(job_id=job_id)
+            query = (
+                select(Revision)
+                .join(Source, Source.id == Revision.source_id)
+                .where(Source.job_id == job_id)
+                .order_by(Revision.date_created)  # pyright: ignore[reportArgumentType]
+            )
             result = await _session.exec(query)
             return result.all()
 
 
-def _transcript_job_query(job_id: UUID):
-    return (
-        select(Transcript)
-        .where(Transcript.job_id == job_id)
-        .options(selectinload(Transcript.job))  # pyright: ignore[reportArgumentType]
-        .order_by(Transcript.revision)  # pyright: ignore[reportArgumentType]
-    )  # fmt: skip
-
-
 def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str:
     provider_model = getattr(provider, "model", None)
     if isinstance(provider_model, str) and provider_model.strip():
diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py
index 9c4a114..c06ffb1 100644
--- a/src/transcription/services/workflows.py
+++ b/src/transcription/services/workflows.py
@@ -10,6 +10,7 @@ from ..errors import classify_unexpected_error
 from ..errors import format_error_detail
 from ..models import Job
 from ..models import JobStatus
+from ..models import Source
 from ..providers import TranscriptionResult
 from . import ServiceBundle
 from .transcription import DEFAULT_PROMPT_FILE
@@ -64,18 +65,17 @@ async def process_queued_job(
         job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session)
         await session.commit()
 
-    document = job.document
-    assert document is not None, (
-        f"Job {job.id} has no associated document or the document failed to be loaded by the job service."
-    )
+    source = _resolve_primary_source(job)
+    assert source is not None, f"Job {job.id} has no associated source record."
 
     try:
-        result = await transcribe_document_image(document.file_path)
+        result = await transcribe_document_image(source.file_path)
         job = await _finalize_transcribed(job=job, services=services, result=result, session=session)
         logger.info(
-            "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s",
+            "Job transcribed operation=worker.process_job job_id=%s document_id=%s source_id=%s provider=%s",
             job.id,
-            document.id,
+            job.document_id,
+            source.id,
             result.provider,
         )
     except Exception as exc:  # noqa: BLE001
@@ -87,9 +87,10 @@ async def process_queued_job(
 
         job = await _finalize_failed(job=job, services=services, error=error, session=session)
         logger.error(
-            "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
+            "Job failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
             job.id,
-            document.id,
+            job.document_id,
+            source.id,
             error.error_id,
             error.category.value,
         )
@@ -118,10 +119,10 @@ async def _finalize_transcribed(
     result: TranscriptionResult,
     session: AsyncSession | None = None,
 ) -> Job:
-    """Transaction B: transcript + TRANSCRIBED in one commit."""
+    """Transaction B: job transcription output + TRANSCRIBED in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.create_transcript_for_job(
+            await services.transcriptions.update_job_transcription(
                 job_id=job.id,
                 text=result.text,
                 error_detail=None,
@@ -138,7 +139,7 @@ async def _finalize_transcribed(
             await local_session.commit()
             return updated_job
 
-    await services.transcriptions.create_transcript_for_job(
+    await services.transcriptions.update_job_transcription(
         job_id=job.id,
         text=result.text,
         error_detail=None,
@@ -164,10 +165,10 @@ async def _finalize_retry(
     settings: Settings,
     session: AsyncSession | None = None,
 ) -> Job:
-    """Transaction C: transcript error + QUEUED + retry increment in one commit."""
+    """Transaction C: job error detail + QUEUED + retry increment in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.create_transcript_for_job(
+            await services.transcriptions.update_job_transcription(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
@@ -182,7 +183,7 @@ async def _finalize_retry(
             )
             await local_session.commit()
     else:
-        await services.transcriptions.create_transcript_for_job(
+        await services.transcriptions.update_job_transcription(
             job_id=job.id,
             text=None,
             error_detail=format_error_detail(error),
@@ -209,10 +210,10 @@ async def _finalize_failed(
     error: AppError,
     session: AsyncSession | None = None,
 ) -> Job:
-    """Transaction B: transcript error + FAILED in one commit."""
+    """Transaction B: job error detail + FAILED in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.create_transcript_for_job(
+            await services.transcriptions.update_job_transcription(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
@@ -227,7 +228,7 @@ async def _finalize_failed(
             await local_session.commit()
             return updated_job
 
-    await services.transcriptions.create_transcript_for_job(
+    await services.transcriptions.update_job_transcription(
         job_id=job.id,
         text=None,
         error_detail=format_error_detail(error),
@@ -241,3 +242,9 @@ async def _finalize_failed(
     )
     await session.commit()
     return updated_job
+
+
+def _resolve_primary_source(job: Job) -> Source | None:
+    if not job.sources:
+        return None
+    return job.sources[0]
diff --git a/src/transcription/ui/components/document_panzoom.py b/src/transcription/ui/components/document_panzoom.py
index bd10965..9d3695c 100644
--- a/src/transcription/ui/components/document_panzoom.py
+++ b/src/transcription/ui/components/document_panzoom.py
@@ -10,24 +10,24 @@ from uuid import uuid4
 from nicegui import ui
 
 from transcription.config import get_settings
-from transcription.models import Document
+from transcription.models import Source
 
 PANGOZOOM_CDN_URL = "https://unpkg.com/@panzoom/panzoom@4.6.2/dist/panzoom.min.js"
 UPLOADS_URL_PREFIX = "/uploads"
 
 
-def render_document_panzoom(*, document: Document) -> None:
-    """Render a document preview with pan and zoom interactions."""
+def render_document_panzoom(*, source: Source) -> None:
+    """Render a source preview with pan and zoom interactions."""
     _register_panzoom_assets()
 
     host_id = f"document-panzoom-{uuid4().hex}"
-    document_url = _document_url(document)
-    document_kind = _document_kind(document)
+    document_url = _document_url(source)
+    document_kind = _document_kind(source)
 
     with ui.card().classes("w-full q-pa-md"):
         with ui.row().classes("w-full items-center justify-between no-wrap"):
             ui.label("Document preview").classes("text-subtitle1 text-weight-medium")
-            ui.label(document.filename).classes("text-caption text-grey-4 ellipsis").style(
+            ui.label(source.filename).classes("text-caption text-grey-4 ellipsis").style(
                 "max-width: 60%; text-align: right;"
             )
 
@@ -40,13 +40,13 @@ def render_document_panzoom(*, document: Document) -> None:
                 if document_kind == "pdf":
                     ui.html(
                         f'<iframe class="document-panzoom-iframe" '
-                        f'src="{document_url}" title="{document.filename}" '
+                        f'src="{document_url}" title="{source.filename}" '
                         "data-panzoom-target></iframe>"
                     )
                 else:
                     ui.html(
                         f'<img class="document-panzoom-media" '
-                        f'src="{document_url}" alt="{document.filename}" '
+                        f'src="{document_url}" alt="{source.filename}" '
                         "data-panzoom-target data-panzoom-media />"
                     )
 
@@ -98,8 +98,8 @@ def _register_panzoom_assets() -> None:
     )
 
 
-def _document_url(document: Document) -> str:
-    file_path = Path(document.file_path)
+def _document_url(source: Source) -> str:
+    file_path = Path(source.file_path)
     upload_dir = get_settings().upload_dir
 
     relative_path: Path
@@ -117,8 +117,8 @@ def _document_url(document: Document) -> str:
     return f"{UPLOADS_URL_PREFIX}/{encoded_relative_path}"
 
 
-def _document_kind(document: Document) -> str:
-    suffix = Path(document.file_path).suffix.lower()
+def _document_kind(source: Source) -> str:
+    suffix = Path(source.file_path).suffix.lower()
     if suffix == ".pdf":
         return "pdf"
     return "image"
diff --git a/src/transcription/ui/components/job_detail.py b/src/transcription/ui/components/job_detail.py
index 70d624a..e7d18e3 100644
--- a/src/transcription/ui/components/job_detail.py
+++ b/src/transcription/ui/components/job_detail.py
@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
 
 from nicegui import ui
 
-from transcription.models import Document
 from transcription.models import Job
-from transcription.models import Transcript
+from transcription.models import Revision
+from transcription.models import Source
 from transcription.ui.components.document_panzoom import render_document_panzoom
+from transcription.ui.components.transcript import render_original_transcription_card
+from transcription.ui.components.transcript import render_revision_row
 
 logger = logging.getLogger(__name__)
 
@@ -33,50 +34,35 @@ def _metadata_row(label: str, value: str) -> None:
         ui.label(value).classes("text-body2 text-right text-grey-1 break-all")
 
 
-def _render_document_section(document: Document) -> None:
+def _render_source_section(source: Source) -> None:
     with ui.card().classes("w-full q-pa-md"):
-        ui.label("Document").classes("text-subtitle1 text-weight-medium")
+        ui.label("Source").classes("text-subtitle1 text-weight-medium")
         ui.separator().classes("q-my-sm")
         with ui.column().classes("w-full q-gutter-y-xs"):
-            _metadata_row("Filename", document.filename)
-            _metadata_row("File path", document.file_path)
+            _metadata_row("Upload name", source.upload_name)
+            _metadata_row("Stored filename", source.filename)
+            _metadata_row("File path", source.file_path)
+            _metadata_row("Uploaded", source.date_uploaded.isoformat())
 
         ui.separator().classes("q-my-md")
-        render_document_panzoom(document=document)
+        render_document_panzoom(source=source)
 
 
-def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
+def _render_revision_section(revision: Revision | None) -> None:
     with ui.card().classes("w-full q-pa-md"):
-        ui.label("Transcripts").classes("text-subtitle1 text-weight-medium")
+        ui.label("Revision").classes("text-subtitle1 text-weight-medium")
         ui.separator().classes("q-my-sm")
 
-        if not transcripts:
-            ui.label("Transcript history is not available yet.").classes("text-body2 text-grey-3")
+        if revision is None:
+            ui.label("No revision exists for this source.").classes("text-body2 text-grey-3")
             return
 
-        for transcript in transcripts:
-            with ui.column().classes("w-full q-gutter-y-xs"):
-                _metadata_row("Revision", str(transcript.revision))
-                _metadata_row("Provider", transcript.provider)
-                _metadata_row("Prompt", transcript.prompt_name)
-                _metadata_row("Created", transcript.created_at.isoformat())
-
-            if transcript.text:
-                ui.separator().classes("q-my-sm")
-                with ui.card().classes("w-fullq-pa-sm"):
-                    ui.markdown(transcript.text).classes("text-grey-1")
-            elif transcript.error_detail:
-                ui.separator().classes("q-my-sm")
-                with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
-                    ui.label("Failure detail").classes("text-caption text-uppercase")
-                    ui.label(transcript.error_detail).classes("text-body2")
-
-            ui.separator().classes("q-my-md bg-blue-grey-7")
+        render_revision_row(revision=revision, initially_expanded=True)
 
 
-def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> None:
+def render_job_detail(*, job: Job, source: Source | None, revision: Revision | None) -> None:
     """Render all sections for the job detail page."""
-    logger.debug("Rendering job detail for job ID %s with %d transcripts", job.id, len(transcripts))
+    logger.debug("Rendering job detail for job ID %s", job.id)
     status_text = job.status.value
     with ui.column().classes("w-full max-w-4xl q-gutter-md"):
         with ui.card().classes("w-full q-pa-lg"):
@@ -93,11 +79,13 @@ def render_job_detail(*, job: Job, document: Document | None, transcripts: Seque
 
             ui.separator().classes("q-my-md bg-blue-grey-7")
             with ui.column().classes("w-full q-gutter-y-xs"):
-                _metadata_row("Created", job.created_at.isoformat())
-                _metadata_row("Updated", job.updated_at.isoformat())
+                _metadata_row("Created", job.date_created.isoformat())
+                _metadata_row("Updated", job.date_updated.isoformat())
                 _metadata_row("Retries", str(job.retry_count))
 
-        if document is not None:
-            _render_document_section(document)
+        render_original_transcription_card(job=job)
 
-        _render_transcript_section(transcripts)
+        if source is not None:
+            _render_source_section(source)
+
+        _render_revision_section(revision)
diff --git a/src/transcription/ui/components/table/jobs.py b/src/transcription/ui/components/table/jobs.py
index a52d786..44f1c22 100644
--- a/src/transcription/ui/components/table/jobs.py
+++ b/src/transcription/ui/components/table/jobs.py
@@ -22,8 +22,8 @@ class JobTableRow:
     status: str
     filename: str
     retry_count: int
-    created_at: str
-    updated_at: str
+    date_created: str
+    date_updated: str
 
 
 def _format_timestamp(value: str) -> str:
@@ -43,10 +43,10 @@ def _serialize_rows(rows: Sequence[JobTableRow]) -> list[dict[str, Any]]:
             "status": row.status,
             "filename": row.filename,
             "retry_count": row.retry_count,
-            "created_at": _format_timestamp(row.created_at),
-            "updated_at": _format_timestamp(row.updated_at),
-            "created_sort": row.created_at,
-            "updated_sort": row.updated_at,
+            "date_created": _format_timestamp(row.date_created),
+            "date_updated": _format_timestamp(row.date_updated),
+            "created_sort": row.date_created,
+            "updated_sort": row.date_updated,
         }
         for row in rows
     ]
@@ -65,8 +65,8 @@ def render_jobs_table(rows: Sequence[JobTableRow]) -> None:
             {"name": "status", "label": "Status", "field": "status", "sortable": True},
             {"name": "filename", "label": "Filename", "field": "filename", "sortable": True},
             {"name": "retry_count", "label": "Retries", "field": "retry_count", "sortable": True},
-            {"name": "created_at", "label": "Created", "field": "created_at", "sortable": True},
-            {"name": "updated_at", "label": "Updated", "field": "updated_at", "sortable": True},
+            {"name": "date_created", "label": "Created", "field": "date_created", "sortable": True},
+            {"name": "date_updated", "label": "Updated", "field": "date_updated", "sortable": True},
         ],
         default_sort_by="created_sort",
         default_descending=True,
diff --git a/src/transcription/ui/components/transcript.py b/src/transcription/ui/components/transcript.py
index 3ec2cad..6ddf3c6 100644
--- a/src/transcription/ui/components/transcript.py
+++ b/src/transcription/ui/components/transcript.py
@@ -9,22 +9,49 @@ from typing import Any
 
 from nicegui import ui
 
-from transcription.models import Transcript
+from transcription.models import Job
+from transcription.models import Revision
 
-type TranscriptAction = Callable[[Transcript], Awaitable[None] | None]
+type RevisionAction = Callable[[Revision], Awaitable[None] | None]
 
 
-def render_transcript_revision_row(
+def render_original_transcription_card(*, job: Job, classes: str = "w-full") -> Any:
+    """Render the immutable original job transcription output."""
+    status_label = "Failed" if job.error_detail else "Transcribed"
+    header = f"Original Transcription | {status_label}"
+    provider = job.provider or "unknown"
+    model = job.model or "unknown"
+    caption = f"{provider} | {model} | {_format_created_at(job.date_updated)}"
+
+    card = ui.card().classes(f"{classes} q-pa-md bg-blue-grey-10")
+    with card, ui.column().classes("w-full q-gutter-y-sm"):
+        ui.label(header).classes("text-subtitle1 text-weight-medium")
+        ui.label(caption).classes("text-caption text-grey-5")
+        _metadata_row(label="Prompt", value=job.prompt_name or "unknown")
+        _metadata_row(label="Updated", value=_format_created_at(job.date_updated))
+
+        if job.text:
+            with ui.card().classes("w-full q-pa-sm"):
+                ui.markdown(job.text)
+
+        if job.error_detail:
+            with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
+                ui.label("Failure detail").classes("text-caption text-uppercase")
+                ui.label(job.error_detail).classes("text-body2")
+
+    return card
+
+
+def render_revision_row(
     *,
-    transcript: Transcript,
+    revision: Revision,
     initially_expanded: bool = False,
     classes: str = "w-full",
-    on_delete: TranscriptAction | None = None,
+    on_delete: RevisionAction | None = None,
 ) -> Any:
-    """Render one collapsible row for a single transcript revision."""
-    status_label = "Failed" if transcript.error_detail else "Transcribed"
-    header = f"Revision {transcript.revision} | {status_label}"
-    caption = f"{transcript.provider} | {transcript.model} | {_format_created_at(transcript.created_at)}"
+    """Render a collapsible row for the single optional source revision."""
+    header = "Revision | User-authored"
+    caption = _format_created_at(revision.date_created)
 
     expansion = ui.expansion(value=initially_expanded, group="group").classes(
         f"{classes} rounded-borders bg-blue-grey-10"
@@ -51,7 +78,7 @@ def render_transcript_revision_row(
                     if not confirmed:
                         return
 
-                    maybe_awaitable = on_delete(transcript)
+                    maybe_awaitable = on_delete(revision)
                     if isinstance(maybe_awaitable, Awaitable):
                         await maybe_awaitable
 
@@ -59,18 +86,11 @@ def render_transcript_revision_row(
                     ui.button(icon="delete", on_click=delete_current_transcript).props(
                         'flat round dense color="negative"'
                     )
-        _metadata_row(label="Provider", value=transcript.provider)
-        _metadata_row(label="Model", value=transcript.model)
-        _metadata_row(label="Created", value=_format_created_at(transcript.created_at))
+        _metadata_row(label="Created", value=_format_created_at(revision.date_created))
 
-        if transcript.text:
+        if revision.text:
             with ui.card().classes("w-full q-pa-sm"):
-                ui.markdown(transcript.text)
-
-        if transcript.error_detail:
-            with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
-                ui.label("Failure detail").classes("text-caption text-uppercase")
-                ui.label(transcript.error_detail).classes("text-body2")
+                ui.markdown(revision.text)
 
     return expansion
 
diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py
index b7afa91..5d446ab 100644
--- a/src/transcription/ui/pages/jobs_page.py
+++ b/src/transcription/ui/pages/jobs_page.py
@@ -8,7 +8,9 @@ from fastapi import Request
 from nicegui import ui
 
 from transcription.app_state import resolve_session_factory
+from transcription.models import Job
 from transcription.models import JobStatus
+from transcription.models import Source
 from transcription.services.jobs import JobService
 from transcription.services.transcription import TranscriptionService
 from transcription.ui.components.app_shell import render_navigation_header
@@ -17,7 +19,8 @@ from transcription.ui.components.table.jobs import render_jobs_table
 
 from ..components.document_panzoom import render_document_panzoom
 from ..components.table.jobs import JobTableRow
-from ..components.transcript import render_transcript_revision_row
+from ..components.transcript import render_original_transcription_card
+from ..components.transcript import render_revision_row
 
 
 def register_page() -> None:
@@ -37,8 +40,8 @@ def register_page() -> None:
                     status=job.status.value,
                     filename=job.filename,
                     retry_count=job.retry_count,
-                    created_at=job.created_at.isoformat(),
-                    updated_at=job.updated_at.isoformat(),
+                    date_created=job.date_created.isoformat(),
+                    date_updated=job.date_updated.isoformat(),
                 )
                 for job in await jobs_service.list_jobs()
             ]
@@ -55,10 +58,14 @@ def register_page() -> None:
         render_navigation_header(current_path="/jobs")
 
         job = await jobs_service.read_job(job_id=UUID(job_id))
+        source = _resolve_primary_source(job)
 
         with ui.splitter(value=30).classes("w-full h-[calc(100vh-64px)]") as splitter:
             with splitter.before, ui.column(align_items="stretch").classes("w-full h-full p-4 gap-3"):
-                render_document_panzoom(document=job.document)
+                if source is not None:
+                    render_document_panzoom(source=source)
+                else:
+                    ui.label("No source preview is available for this job.").classes("text-body2 text-grey-3")
             with splitter.after, ui.column(align_items="stretch").classes("w-full h-full p-4 gap-3"):
                 with ui.row():
                     ui.button(icon="arrow_back", on_click=ui.navigate.back)
@@ -70,30 +77,37 @@ def register_page() -> None:
                         case _:
                             ui.label(f"{job.status.value}").classes("text-subtitle1 text-weight-medium")
 
-                async def delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
+                render_original_transcription_card(job=job)
+
+                async def delete_revision_by_id(revision_id: UUID) -> None:
                     try:
-                        transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
-                        await transcription_service.delete_transcript(transcript)
+                        revision = await transcription_service.read_revision(revision_id=revision_id)
+                        await transcription_service.delete_revision(revision)
                     except Exception as exc:  # noqa: BLE001
-                        show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
+                        show_error(exc, title="Delete failed", operation="jobs.delete_revision")
                         return
 
-                    ui.notify(f"Deleted revision {revision}", type="positive")
-                    await render_transcript_list.refresh()
+                    ui.notify("Deleted revision", type="positive")
+                    await render_revision_panel.refresh()
 
                 @ui.refreshable
-                async def render_transcript_list() -> None:
+                async def render_revision_panel() -> None:
                     refreshed_job = await jobs_service.read_job(job_id=UUID(job_id))
-                    for i, transcript in enumerate(refreshed_job.transcripts):
-                        render_transcript_revision_row(
-                            transcript=transcript,
-                            initially_expanded=(i == 0),
-                            on_delete=(
-                                lambda _transcript, tid=transcript.id, rev=transcript.revision: delete_transcript_by_id(
-                                    tid,
-                                    rev,
-                                )
-                            ),
-                        )
+                    refreshed_source = _resolve_primary_source(refreshed_job)
+                    if refreshed_source is None or refreshed_source.revision is None:
+                        ui.label("No revision exists for this source.").classes("text-body2 text-grey-3")
+                        return
 
-                await render_transcript_list()
+                    render_revision_row(
+                        revision=refreshed_source.revision,
+                        initially_expanded=True,
+                        on_delete=lambda _revision, rid=refreshed_source.revision.id: delete_revision_by_id(rid),
+                    )
+
+                await render_revision_panel()
+
+
+def _resolve_primary_source(job: Job) -> Source | None:
+    if not job.sources:
+        return None
+    return job.sources[0]
diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py
index 4e83315..481f893 100644
--- a/tests/integration/test_pipeline_flow.py
+++ b/tests/integration/test_pipeline_flow.py
@@ -3,12 +3,12 @@
 from pathlib import Path
 
 import pytest
-from sqlmodel import select
 
 from transcription.config import Settings
-from transcription.models import Job, JobStatus, Transcript
+from transcription.models import Job
+from transcription.models import JobStatus
 from transcription.providers.base import TranscriptionResult
-from transcription.services.upload import create_upload_job
+from transcription.services.store import create_upload_job
 from transcription.worker import process_next_queued_job
 
 
@@ -16,61 +16,64 @@ from transcription.worker import process_next_queued_job
 class TestPipelineSuccessFlow:
     """Verify end-to-end success lifecycle behavior."""
 
-    def test_upload_then_worker_persists_transcribed_terminal_state(self, session, tmp_path: Path, monkeypatch):
-        """Upload followed by worker processing persists transcript and transcribed status."""
+    @pytest.mark.asyncio
+    async def test_upload_then_worker_persists_transcribed_terminal_state(
+        self, async_session, tmp_path: Path, monkeypatch
+    ):
+        """Upload followed by worker processing persists job transcription and transcribed status."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        upload_result = create_upload_job(
+        upload_result = await create_upload_job(
             filename="pipeline.jpg",
             file_bytes=b"pipeline-bytes",
-            session=session,
+            session=async_session,
             settings=settings,
         )
 
-        def _fake_transcribe(_path: str) -> TranscriptionResult:
-            return TranscriptionResult(text="Pipeline transcript", provider="openrouter", model="test-model")
+        async def _fake_transcribe(*, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+            _ = (prompt_text, image_bytes, mime_type)
+            return TranscriptionResult(text="Pipeline transcript", provider="openrouter", model="test-model", prompt_name="transcribe_document.md")
 
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
+        monkeypatch.setattr("transcription.services.transcription.OpenRouterTranscriptionProvider.transcribe", _fake_transcribe)
 
-        processed = process_next_queued_job(session=session)
-        job = session.get(Job, upload_result.job_id)
-        transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
+        processed = await process_next_queued_job(session=async_session)
+        job = await async_session.get(Job, upload_result.job_id)
 
         assert processed is True
         assert job is not None
         assert job.status == JobStatus.TRANSCRIBED
-        assert transcript is not None
-        assert transcript.text == "Pipeline transcript"
-        assert transcript.error_detail is None
+        assert job.text == "Pipeline transcript"
+        assert job.error_detail is None
 
 
 @pytest.mark.integration
 class TestPipelineFailureFlow:
     """Verify end-to-end failure lifecycle behavior."""
 
-    def test_upload_then_worker_persists_failed_terminal_state(self, session, tmp_path: Path, monkeypatch):
-        """Upload followed by worker processing persists error detail and failed status."""
+    @pytest.mark.asyncio
+    async def test_upload_then_worker_persists_failed_terminal_state(self, async_session, tmp_path: Path, monkeypatch):
+        """Upload followed by worker processing persists error detail and failed status on the job."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        upload_result = create_upload_job(
+        upload_result = await create_upload_job(
             filename="pipeline.jpg",
             file_bytes=b"pipeline-bytes",
-            session=session,
+            session=async_session,
             settings=settings,
         )
 
-        def _fake_transcribe(_path: str) -> TranscriptionResult:
+        async def _fake_transcribe(*, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+            _ = (prompt_text, image_bytes, mime_type)
             raise RuntimeError("pipeline provider failure")
 
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
+        monkeypatch.setattr("transcription.services.transcription.OpenRouterTranscriptionProvider.transcribe", _fake_transcribe)
 
-        processed = process_next_queued_job(session=session)
-        job = session.get(Job, upload_result.job_id)
-        transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
+        processed = await process_next_queued_job(session=async_session)
+        job = await async_session.get(Job, upload_result.job_id)
 
         assert processed is True
         assert job is not None
         assert job.status == JobStatus.FAILED
-        assert transcript is not None
-        assert transcript.text is None
-        assert "pipeline provider failure" in transcript.error_detail
-        assert "[internal_unexpected_error]" in transcript.error_detail
-        assert "error_id=" in transcript.error_detail
+        assert job.text is None
+        assert job.error_detail is not None
+        assert "pipeline provider failure" in job.error_detail
+        assert "[internal_unexpected_error]" in job.error_detail
+        assert "error_id=" in job.error_detail
diff --git a/tests/test_db.py b/tests/test_db.py
index d6d615f..8e069f0 100644
--- a/tests/test_db.py
+++ b/tests/test_db.py
@@ -18,10 +18,10 @@ class TestSchemaBootstrap:
     """Verify create_all produces the expected table set."""
 
     def test_create_all_creates_expected_tables(self):
-        """After create_all(), document, job, and transcript tables exist."""
+        """After create_all(), document, source, job, and revision tables exist."""
         engine = _in_memory_engine()
         # Ensure models are imported so metadata is populated
-        from transcription.models import Document, Job, Transcript  # noqa: F401
+        from transcription.models import Document, Job, Revision, Source  # noqa: F401
 
         import transcription.db as db_module
 
@@ -31,7 +31,8 @@ class TestSchemaBootstrap:
         table_names = set(inspector.get_table_names())
         assert "document" in table_names
         assert "job" in table_names
-        assert "transcript" in table_names
+        assert "source" in table_names
+        assert "revision" in table_names
 
 
 class TestSessionFactory:
diff --git a/tests/test_models.py b/tests/test_models.py
index d2b710b..b85ad8f 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -1,31 +1,28 @@
-"""Tests for transcription.models — Document, Job, Transcript persistence and relationships."""
+"""Tests for transcription.models — Document, Source, Job, Revision persistence and relationships."""
 
 from uuid import UUID
 
 import pytest
 from sqlalchemy.exc import IntegrityError
 
-from transcription.models import Document, Job, JobStatus, Transcript
+from transcription.models import Document, Job, JobStatus, Revision, Source
 
 
 def _make_document(**overrides) -> Document:
-    """Create a Document with sensible defaults."""
-    defaults = {"filename": "letter.jpg", "file_path": "/uploads/letter.jpg"}
+    defaults = {"name": "letter bundle"}
     defaults.update(overrides)
     return Document(**defaults)
 
 
 def _persist_document(session) -> Document:
-    """Create, persist, and return a Document."""
-    doc = _make_document()
-    session.add(doc)
+    document = _make_document()
+    session.add(document)
     session.commit()
-    session.refresh(doc)
-    return doc
+    session.refresh(document)
+    return document
 
 
 def _persist_job(session, document: Document) -> Job:
-    """Create, persist, and return a Job linked to a Document."""
     job = Job(document_id=document.id)
     session.add(job)
     session.commit()
@@ -33,147 +30,132 @@ def _persist_job(session, document: Document) -> Job:
     return job
 
 
-class TestDocumentModel:
-    """Verify Document creation and default field population."""
+def _persist_source(session, document: Document, job: Job, **overrides) -> Source:
+    defaults = {
+        "document_id": document.id,
+        "job_id": job.id,
+        "upload_name": "letter.jpg",
+        "filename": "stored-letter.jpg",
+        "file_path": "/uploads/stored-letter.jpg",
+    }
+    defaults.update(overrides)
+    source = Source(**defaults)
+    session.add(source)
+    session.commit()
+    session.refresh(source)
+    return source
 
+
+class TestDocumentModel:
     def test_can_be_persisted(self, session):
-        """A Document round-trips through the database with correct fields."""
-        doc = _persist_document(session)
-        fetched = session.get(Document, doc.id)
+        document = _persist_document(session)
+        fetched = session.get(Document, document.id)
         assert fetched is not None
-        assert fetched.filename == "letter.jpg"
-        assert fetched.file_path == "/uploads/letter.jpg"
+        assert fetched.name == "letter bundle"
 
     def test_defaults_are_populated(self, session):
-        """id is a UUID and uploaded_at is populated on creation."""
-        doc = _persist_document(session)
-        assert isinstance(doc.id, UUID)
-        assert doc.uploaded_at is not None
+        document = _persist_document(session)
+        assert isinstance(document.id, UUID)
 
 
 class TestJobModel:
-    """Verify Job creation, defaults, and status transitions."""
-
     def test_can_be_created_for_document(self, session):
-        """A Job linked to a Document via FK persists correctly."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
+        document = _persist_document(session)
+        job = _persist_job(session, document)
         fetched = session.get(Job, job.id)
         assert fetched is not None
-        assert fetched.document_id == doc.id
+        assert fetched.document_id == document.id
 
     def test_defaults_are_populated(self, session):
-        """Default status is queued; created_at and updated_at are populated."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
+        document = _persist_document(session)
+        job = _persist_job(session, document)
         assert job.status == JobStatus.QUEUED
         assert job.retry_count == 0
-        assert job.created_at is not None
-        assert job.updated_at is not None
+        assert job.date_created is not None
+        assert job.date_updated is not None
 
     def test_transitions_to_transcribed(self, session):
-        """Status updates from queued to processing to transcribed."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        assert job.status == JobStatus.QUEUED
+        document = _persist_document(session)
+        job = _persist_job(session, document)
 
         job.status = JobStatus.PROCESSING
         session.add(job)
         session.commit()
         session.refresh(job)
-        assert job.status == JobStatus.PROCESSING
 
         job.status = JobStatus.TRANSCRIBED
         session.add(job)
         session.commit()
         session.refresh(job)
+
         assert job.status == JobStatus.TRANSCRIBED
 
-    def test_transitions_to_failed(self, session):
-        """Status updates from processing to failed."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
 
-        job.status = JobStatus.PROCESSING
-        session.add(job)
-        session.commit()
-        session.refresh(job)
+class TestSourceModel:
+    def test_can_be_created_for_document_and_job(self, session):
+        document = _persist_document(session)
+        job = _persist_job(session, document)
+        source = _persist_source(session, document, job)
 
-        job.status = JobStatus.FAILED
-        session.add(job)
-        session.commit()
-        session.refresh(job)
-        assert job.status == JobStatus.FAILED
-
-
-class TestTranscriptModel:
-    """Verify Transcript persistence for success and failure cases."""
-
-    def test_success_record_persists(self, session):
-        """A Transcript with text set and error_detail None persists correctly."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        transcript = Transcript(job_id=job.id, text="Dear Sir, ...")
-        session.add(transcript)
-        session.commit()
-        session.refresh(transcript)
-
-        fetched = session.get(Transcript, transcript.id)
+        fetched = session.get(Source, source.id)
         assert fetched is not None
-        assert fetched.text == "Dear Sir, ..."
-        assert fetched.error_detail is None
+        assert fetched.document_id == document.id
+        assert fetched.job_id == job.id
+        assert fetched.date_uploaded is not None
 
-    def test_failure_record_persists(self, session):
-        """A Transcript with text None and error_detail set persists correctly."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        transcript = Transcript(job_id=job.id, error_detail="Provider timeout")
-        session.add(transcript)
+
+class TestRevisionModel:
+    def test_revision_persists_for_source(self, session):
+        document = _persist_document(session)
+        job = _persist_job(session, document)
+        source = _persist_source(session, document, job)
+
+        revision = Revision(source_id=source.id, text="Edited revision text")
+        session.add(revision)
         session.commit()
-        session.refresh(transcript)
+        session.refresh(revision)
 
-        fetched = session.get(Transcript, transcript.id)
+        fetched = session.get(Revision, revision.id)
         assert fetched is not None
-        assert fetched.text is None
-        assert fetched.error_detail == "Provider timeout"
+        assert fetched.text == "Edited revision text"
+        assert fetched.date_created is not None
 
-    def test_job_id_is_unique(self, session):
-        """Inserting two transcripts with the same job_id raises an integrity error."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
+    def test_source_id_is_unique(self, session):
+        document = _persist_document(session)
+        job = _persist_job(session, document)
+        source = _persist_source(session, document, job)
 
-        t1 = Transcript(job_id=job.id, text="First")
-        session.add(t1)
+        first = Revision(source_id=source.id, text="First")
+        session.add(first)
         session.commit()
 
-        t2 = Transcript(job_id=job.id, text="Duplicate")
-        session.add(t2)
+        duplicate = Revision(source_id=source.id, text="Duplicate")
+        session.add(duplicate)
         with pytest.raises(IntegrityError):
             session.commit()
 
 
 class TestRelationships:
-    """Verify SQLModel relationship navigation between models."""
+    def test_document_exposes_jobs_and_sources(self, session):
+        document = _persist_document(session)
+        job = _persist_job(session, document)
+        _persist_source(session, document, job)
 
-    def test_document_exposes_jobs(self, session):
-        """document.jobs returns the linked Job list."""
-        doc = _persist_document(session)
-        _persist_job(session, doc)
-        _persist_job(session, doc)
+        session.refresh(document)
+        assert len(document.jobs) == 1
+        assert len(document.sources) == 1
 
-        session.refresh(doc)
-        assert len(doc.jobs) == 2
-        assert all(isinstance(j, Job) for j in doc.jobs)
+    def test_source_exposes_optional_single_revision(self, session):
+        document = _persist_document(session)
+        job = _persist_job(session, document)
+        source = _persist_source(session, document, job)
 
-    def test_job_exposes_transcript(self, session):
-        """job.transcript returns the linked Transcript."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        transcript = Transcript(job_id=job.id, text="Transcribed text")
-        session.add(transcript)
+        assert source.revision is None
+
+        revision = Revision(source_id=source.id, text="Edited")
+        session.add(revision)
         session.commit()
 
-        session.refresh(job)
-        assert job.transcript is not None
-        assert isinstance(job.transcript, Transcript)
-        assert job.transcript.text == "Transcribed text"
+        session.refresh(source)
+        assert source.revision is not None
+        assert source.revision.text == "Edited"
diff --git a/uploads/287bb88d-80c1-4553-aedc-35f2167c49b6_Book Two - page 61.jpg b/uploads/287bb88d-80c1-4553-aedc-35f2167c49b6_Book Two - page 61.jpg
deleted file mode 100644
index de4eac7..0000000
Binary files a/uploads/287bb88d-80c1-4553-aedc-35f2167c49b6_Book Two - page 61.jpg and /dev/null differ
diff --git a/uploads/7c6ee648-4171-4e70-b901-ba94ee227d6a_Time Rolls On - page 013.jpg b/uploads/7c6ee648-4171-4e70-b901-ba94ee227d6a_Time Rolls On - page 013.jpg
deleted file mode 100644
index 5f74c67..0000000
Binary files a/uploads/7c6ee648-4171-4e70-b901-ba94ee227d6a_Time Rolls On - page 013.jpg and /dev/null differ
diff --git a/uploads/b04b640b-dce2-4543-b8da-74ea764bdae7_Time Rolls On - page 011.jpg b/uploads/b04b640b-dce2-4543-b8da-74ea764bdae7_Time Rolls On - page 011.jpg
deleted file mode 100644
index 693cae5..0000000
Binary files a/uploads/b04b640b-dce2-4543-b8da-74ea764bdae7_Time Rolls On - page 011.jpg and /dev/null differ
diff --git a/uploads/f2984991-e861-4db3-8612-2da1a8ba23d2_Book Two - page 03.jpg b/uploads/f2984991-e861-4db3-8612-2da1a8ba23d2_Book Two - page 03.jpg
deleted file mode 100644
index e0db646..0000000
Binary files a/uploads/f2984991-e861-4db3-8612-2da1a8ba23d2_Book Two - page 03.jpg and /dev/null differ