delete button on transcript header

.github/instructions/ui.instructions.md

@@ -0,0 +1,6 @@
+---
+description: Copilot rules for modifying the UI
+applyTo: 'src/transcription/ui/**/*.py'
+---
+
+The UI is forbidden from directly touching the database. All interactions should be done using the [services](../../src/transcription/services/)

.gitignore

@@ -14,3 +14,8 @@ wheels/
 
 # SQLite database
 *.db
+
+upload/
+*.jpg
+*.jpeg
+*.png

src/transcription/app.py

@@ -8,6 +8,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi import status
 from fastapi.responses import RedirectResponse
+from fastapi.staticfiles import StaticFiles
 
 from .api.errors import register_error_handlers
 from .api.health import router as health_router
@@ -25,7 +26,7 @@ from .worker import worker_consumer_lifespan
 async def _lifespan(app: FastAPI):
     configure_logging()
 
-    settings = get_settings()
+    settings = getattr(app.state, "settings", None) or get_settings()
     app.state.settings = settings
     app.state.services = ServiceBundle()
     app.state.runtime = initialize_database_runtime(settings=settings)
@@ -52,6 +53,13 @@ async def _lifespan(app: FastAPI):
 def create_app() -> FastAPI:
     """Create and configure the FastAPI application."""
     app = FastAPI(title="Transcription", lifespan=_lifespan)
+    settings = get_settings()
+    app.state.settings = settings
+    app.mount(
+        "/uploads",
+        StaticFiles(directory=settings.upload_dir, check_dir=False),
+        name="uploads",
+    )
 
     @app.get("/", include_in_schema=False)
     async def root_redirect() -> RedirectResponse:

src/transcription/db/operations.py

@@ -51,10 +51,15 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
 
     inspector = inspect(connection)
     table_names = set(inspector.get_table_names())
-    if "job" not in table_names:
-        return
 
-    columns = {column["name"] for column in inspector.get_columns("job")}
-    if "retry_count" not in columns:
-        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 "job" in table_names:
+        job_columns = {column["name"] for column in inspector.get_columns("job")}
+        if "retry_count" not in job_columns:
+            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")

src/transcription/models.py

@@ -1,16 +1,16 @@
 """SQLModel domain models for the transcription system.
 
 Three models capture the MVP lifecycle:
-  Document  ->  one-to-many  ->  Job  ->  one-to-one  ->  Transcript
+    Document  ->  one-to-many  ->  Job  ->  one-to-many  ->  Transcript
 """
 
 from datetime import UTC
 from datetime import datetime
 from enum import StrEnum
-from typing import Optional
 from uuid import UUID
 from uuid import uuid4
 
+from sqlalchemy import UniqueConstraint
 from sqlmodel import Field
 from sqlmodel import Relationship
 from sqlmodel import SQLModel
@@ -47,7 +47,7 @@ class Job(SQLModel, table=True):
 
     # --- relationships ---
     document: Document = Relationship(back_populates="jobs")
-    transcript: Optional["Transcript"] = Relationship(back_populates="job")
+    transcripts: list["Transcript"] = Relationship(back_populates="job")
 
     @property
     def filename(self) -> str:
@@ -59,10 +59,14 @@ class Transcript(SQLModel, table=True):
     """The output of a transcription job."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    job_id: UUID = Field(foreign_key="job.id", unique=True)
-    """ID for the associated job. There's a 1-1 relationship bewteen transcripts and jobs."""
+    job_id: UUID = Field(foreign_key="job.id")
+    """ID for the associated job."""
+    revision: int = Field(default=0, ge=0)
+    """Revision number for this job's transcript history, starting at 0."""
     provider: str
     """Name of the transcription provider used to generate this transcript."""
+    model: str
+    """Model identifier used to generate this transcript revision."""
     prompt_name: str
     """Name of the prompt used to generate this transcript."""
     text: str | None = None
@@ -71,5 +75,7 @@ class Transcript(SQLModel, table=True):
     """Details of any error that occurred during transcription."""
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
+    __table_args__ = (UniqueConstraint("job_id", "revision", name="uq_transcript_job_revision"),)
+
     # --- relationships ---
-    job: Job = Relationship(back_populates="transcript")
+    job: Job = Relationship(back_populates="transcripts")

src/transcription/providers/base.py

@@ -28,12 +28,14 @@ class TranscriptionResult:
     prompt_name: str
     model: str
 
-    def to_transcript(self, job_id: UUID) -> Transcript:
+    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,
         )
 

src/transcription/services/jobs.py

@@ -35,7 +35,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.transcripts),  # pyright: ignore[reportArgumentType]
+                )
                 .where(Job.id == job_id)
                 .execution_options(populate_existing=True)
             )

src/transcription/services/transcription.py

@@ -4,10 +4,12 @@ from __future__ import annotations
 
 import logging
 import mimetypes
+from collections.abc import Sequence
 from contextlib import contextmanager
 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
@@ -108,38 +110,90 @@ class TranscriptionService(ServiceBase):
             settings=self.settings,
             provider=self.provider,
         )
-        await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session)
+        await self.create_transcript_for_job(
+            job_id=job_id,
+            text=result.text,
+            provider=result.provider,
+            model=result.model,
+            prompt_name=result.prompt_name,
+            session=session,
+        )
 
-    async def upsert_transcript_by_job(
+    async def create_transcript_for_job(
         self,
         *,
         job_id: UUID,
         text: str | None,
-        error_detail: str | None,
+        error_detail: str | None = None,
         provider: str | None = None,
+        model: str | None = None,
         prompt_name: str = DEFAULT_PROMPT_FILE,
         session: AsyncSession | None = None,
     ) -> Transcript:
-        """Create or update a transcript for a job id."""
+        """Create a new transcript revision for a job id."""
         async with self._session_scope(session) as _session:
-            transcript = (await _session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
-            if transcript is None:
-                transcript = Transcript(
-                    job_id=job_id,
-                    provider=provider or self.settings.provider.value,
-                    prompt_name=prompt_name,
-                )
-
-            transcript.text = text
-            transcript.error_detail = error_detail
-            if provider is not None:
-                transcript.provider = provider
-            transcript.prompt_name = prompt_name
+            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
 
+            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
 
+    async def read_latest_transcript_by_job(
+        self,
+        job_id: UUID,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Transcript | None:
+        """Read the latest transcript revision for a job id."""
+        async with self._session_scope(session) as _session:
+            query = _transcript_job_query(job_id=job_id).limit(1)
+            result = await _session.exec(query)
+            return result.one_or_none()
+
+    async def list_transcripts_by_job(
+        self,
+        job_id: UUID,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Transcript]:
+        """List transcript revisions for a job id in ascending revision order."""
+        async with self._session_scope(session) as _session:
+            query = _transcript_job_query(job_id=job_id)
+            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():
+        return provider_model
+
+    if settings.provider_model and settings.provider_model.strip():
+        return settings.provider_model
+
+    return "unknown"
+
 
 async def transcribe_document_image(
     image_path: str | Path,

src/transcription/services/workflows.py

@@ -121,11 +121,12 @@ async def _finalize_transcribed(
     """Transaction B: transcript + TRANSCRIBED in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.upsert_transcript_by_job(
+            await services.transcriptions.create_transcript_for_job(
                 job_id=job.id,
                 text=result.text,
                 error_detail=None,
                 provider=result.provider,
+                model=result.model,
                 prompt_name=result.prompt_name,
                 session=local_session,
             )
@@ -137,11 +138,12 @@ async def _finalize_transcribed(
             await local_session.commit()
             return updated_job
 
-    await services.transcriptions.upsert_transcript_by_job(
+    await services.transcriptions.create_transcript_for_job(
         job_id=job.id,
         text=result.text,
         error_detail=None,
         provider=result.provider,
+        model=result.model,
         prompt_name=result.prompt_name,
         session=session,
     )
@@ -165,7 +167,7 @@ async def _finalize_retry(
     """Transaction C: transcript error + QUEUED + retry increment in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.upsert_transcript_by_job(
+            await services.transcriptions.create_transcript_for_job(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
@@ -180,7 +182,7 @@ async def _finalize_retry(
             )
             await local_session.commit()
     else:
-        await services.transcriptions.upsert_transcript_by_job(
+        await services.transcriptions.create_transcript_for_job(
             job_id=job.id,
             text=None,
             error_detail=format_error_detail(error),
@@ -210,7 +212,7 @@ async def _finalize_failed(
     """Transaction B: transcript error + FAILED in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            await services.transcriptions.upsert_transcript_by_job(
+            await services.transcriptions.create_transcript_for_job(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
@@ -225,7 +227,7 @@ async def _finalize_failed(
             await local_session.commit()
             return updated_job
 
-    await services.transcriptions.upsert_transcript_by_job(
+    await services.transcriptions.create_transcript_for_job(
         job_id=job.id,
         text=None,
         error_detail=format_error_detail(error),

src/transcription/ui/components/__init__.py

@@ -2,5 +2,6 @@
 
 from transcription.ui.components.app_shell import NAV_ITEMS
 from transcription.ui.components.app_shell import render_navigation_header
+from transcription.ui.components.document_panzoom import render_document_panzoom
 
-__all__ = ["NAV_ITEMS", "render_navigation_header"]
+__all__ = ["NAV_ITEMS", "render_document_panzoom", "render_navigation_header"]

src/transcription/ui/components/document_panzoom.py

@@ -0,0 +1,211 @@
+"""Panzoom-backed document preview component."""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from urllib.parse import quote
+from uuid import uuid4
+
+from nicegui import ui
+
+from transcription.config import get_settings
+from transcription.models import Document
+
+PANGOZOOM_CDN_URL = "https://unpkg.com/@panzoom/[email protected]/dist/panzoom.min.js"
+UPLOADS_URL_PREFIX = "/uploads"
+
+
+def render_document_panzoom(*, document: Document) -> None:
+    """Render a document 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)
+
+    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(
+                "max-width: 60%; text-align: right;"
+            )
+
+        with (
+            ui.element("div").classes("w-full document-panzoom-host rounded-borders q-mt-md")
+            # .style(f"height: {height};")
+        ) as host:
+            host.props(f"id={host_id}")
+            with ui.element("div").classes("document-panzoom-surface"):
+                if document_kind == "pdf":
+                    ui.html(
+                        f'<iframe class="document-panzoom-iframe" '
+                        f'src="{document_url}" title="{document.filename}" '
+                        "data-panzoom-target></iframe>"
+                    )
+                else:
+                    ui.html(
+                        f'<img class="document-panzoom-media" '
+                        f'src="{document_url}" alt="{document.filename}" '
+                        "data-panzoom-target data-panzoom-media />"
+                    )
+
+    _attach_panzoom(host_id)
+
+
+@lru_cache(maxsize=1)
+def _register_panzoom_assets() -> None:
+    ui.add_head_html(
+        f'<script src="{PANGOZOOM_CDN_URL}"></script>',
+        shared=True,
+    )
+    ui.add_head_html(
+        """
+        <style>
+            .document-panzoom-host {
+                overflow: hidden;
+                touch-action: none;
+            }
+
+            .document-panzoom-surface {
+                width: 100%;
+                height: 100%;
+                display: flex;
+                align-items: flex-start;
+                justify-content: flex-start;
+            }
+
+            .document-panzoom-media {
+                width: auto;
+                height: auto;
+                display: block;
+                max-width: 100%;
+                max-height: 100%;
+                user-select: none;
+                -webkit-user-drag: none;
+            }
+
+            .document-panzoom-iframe {
+                width: 100%;
+                height: 100%;
+                border: 0;
+                pointer-events: none;
+                background: white;
+            }
+        </style>
+        """,
+        shared=True,
+    )
+
+
+def _document_url(document: Document) -> str:
+    file_path = Path(document.file_path)
+    upload_dir = get_settings().upload_dir
+
+    relative_path: Path
+    try:
+        relative_path = file_path.resolve().relative_to(upload_dir.resolve())
+    except ValueError:
+        parts = file_path.parts
+        if "uploads" in parts:
+            uploads_index = parts.index("uploads")
+            relative_path = Path(*parts[uploads_index + 1 :])
+        else:
+            relative_path = Path(file_path.name)
+
+    encoded_relative_path = "/".join(quote(part) for part in relative_path.parts)
+    return f"{UPLOADS_URL_PREFIX}/{encoded_relative_path}"
+
+
+def _document_kind(document: Document) -> str:
+    suffix = Path(document.file_path).suffix.lower()
+    if suffix == ".pdf":
+        return "pdf"
+    return "image"
+
+
+def _attach_panzoom(host_id: str) -> None:
+    ui.run_javascript(
+        f"""
+        (function() {{
+            if (!window.Panzoom) return;
+            window.__transcriptionPanzoom = window.__transcriptionPanzoom || {{}};
+            const host = document.getElementById({host_id!r});
+            if (!host) return;
+            const target = host.querySelector('[data-panzoom-target]');
+            const media = host.querySelector('[data-panzoom-media]');
+            if (!target) return;
+
+            const cleanup = () => {{
+                const existing = window.__transcriptionPanzoom[{host_id!r}];
+                if (existing?.resizeObserver) existing.resizeObserver.disconnect();
+                if (existing?.wheelHandler) host.removeEventListener('wheel', existing.wheelHandler);
+                if (existing?.instance) existing.instance.destroy();
+            }};
+
+            const computeFitScale = () => {{
+                const hostRect = host.getBoundingClientRect();
+                if (hostRect.width <= 0 || hostRect.height <= 0) return null;
+                return 1;
+            }};
+
+            const buildInstance = () => {{
+                cleanup();
+
+                const fitScale = computeFitScale();
+                if (fitScale === null) return false;
+
+                const minScale = Math.min(fitScale, 0.01);
+                const instance = Panzoom(target, {{
+                    startX: 0,
+                    startY: 0,
+                    startScale: fitScale,
+                    minScale: minScale,
+                    maxScale: 256,
+                    step: 0.2,
+                    roundPixels: false,
+                    panOnlyWhenZoomed: true,
+                    overflow: 'hidden',
+                }});
+
+                const wheelHandler = (event) => instance.zoomWithWheel(event);
+                host.addEventListener('wheel', wheelHandler, {{ passive: false }});
+
+                requestAnimationFrame(() => {{
+                    instance.reset({{ animate: false }});
+                }});
+
+                const resizeObserver = new ResizeObserver(() => {{
+                    const nextFitScale = computeFitScale();
+                    if (nextFitScale === null) return;
+                    instance.setOptions({{
+                        startScale: nextFitScale,
+                        minScale: Math.min(nextFitScale, 0.01),
+                    }});
+                    instance.reset({{ animate: false }});
+                }});
+                resizeObserver.observe(host);
+
+                window.__transcriptionPanzoom[{host_id!r}] = {{
+                    instance,
+                    wheelHandler,
+                    resizeObserver,
+                }};
+                return true;
+            }};
+
+            const initWhenReady = (retries = 15) => {{
+                if (buildInstance()) return;
+                if (retries <= 0) return;
+                requestAnimationFrame(() => initWhenReady(retries - 1));
+            }};
+
+            if (media && media.tagName === 'IMG' && !media.complete) {{
+                media.addEventListener('load', () => initWhenReady(), {{ once: true }});
+                return;
+            }}
+
+            initWhenReady();
+        }})();
+        """
+    )

src/transcription/ui/components/job_detail.py

@@ -2,11 +2,17 @@
 
 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.ui.components.document_panzoom import render_document_panzoom
+
+logger = logging.getLogger(__name__)
 
 
 def _status_chip_classes(status: str) -> str:
@@ -28,46 +34,52 @@ def _metadata_row(label: str, value: str) -> None:
 
 
 def _render_document_section(document: Document) -> None:
-    with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 q-pa-md"):
+    with ui.card().classes("w-full q-pa-md"):
         ui.label("Document").classes("text-subtitle1 text-weight-medium")
-        ui.separator().classes("q-my-sm bg-blue-grey-7")
+        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)
 
+        ui.separator().classes("q-my-md")
+        render_document_panzoom(document=document)
 
-def _render_transcript_section(transcript: Transcript | None) -> None:
-    with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 q-pa-md"):
-        ui.label("Transcript").classes("text-subtitle1 text-weight-medium")
-        ui.separator().classes("q-my-sm bg-blue-grey-7")
 
-        if transcript is None:
-            ui.label("Transcript not available yet.").classes("text-body2 text-grey-3")
+def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
+    with ui.card().classes("w-full q-pa-md"):
+        ui.label("Transcripts").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")
             return
 
-        with ui.column().classes("w-full q-gutter-y-xs"):
-            _metadata_row("Provider", transcript.provider)
-            _metadata_row("Prompt", transcript.prompt_name)
-            _metadata_row("Created", transcript.created_at.isoformat())
+        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 bg-blue-grey-7")
-            with ui.card().classes("w-full bg-blue-grey-9 text-grey-1 q-pa-sm"):
-                ui.markdown(transcript.text).classes("text-grey-1")
-            return
+            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")
 
-        if transcript.error_detail:
-            ui.separator().classes("q-my-sm bg-blue-grey-7")
-            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")
 
 
-def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None:
+def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> 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))
     status_text = job.status.value
     with ui.column().classes("w-full max-w-4xl q-gutter-md"):
-        with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 q-pa-lg"):
+        with ui.card().classes("w-full q-pa-lg"):
             with ui.row().classes("w-full items-center justify-between q-gutter-md"):
                 with ui.column().classes("q-gutter-none"):
                     ui.label("Job overview").classes("text-h6 text-weight-bold")
@@ -88,4 +100,4 @@ def render_job_detail(*, job: Job, document: Document | None, transcript: Transc
         if document is not None:
             _render_document_section(document)
 
-        _render_transcript_section(transcript)
+        _render_transcript_section(transcripts)

src/transcription/ui/components/table/common.py

@@ -1,11 +1,14 @@
 """Common logic for generating table widgets."""
 
+import logging
 from collections.abc import Callable
 from typing import Any
 
 from nicegui import events
 from nicegui import ui
 
+logger = logging.getLogger(__name__)
+
 
 def _extract_row_id(args: Any) -> str | None:
     if isinstance(args, dict):
@@ -37,6 +40,7 @@ def _bind_row_click_handler(
         on_row_click_id(row_id)
 
     table.on("rowClick", handle_row_click)
+    logger.debug("Row click handler bound to table")
 
 
 def build_table(
@@ -63,6 +67,7 @@ def build_table(
         .classes(classes)
         .props('table-style="table-layout: fixed; width: 100%;"')
     )
+    logger.info("Table built with %d rows and %d columns", len(rows), len(columns))
     if on_row_click_id is not None:
         _bind_row_click_handler(table, on_row_click_id=on_row_click_id)
     return table

src/transcription/ui/components/transcript.py

@@ -0,0 +1,86 @@
+"""Reusable transcript UI components."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable
+from collections.abc import Callable
+from datetime import datetime
+from typing import Any
+
+from nicegui import ui
+
+from transcription.models import Transcript
+
+type TranscriptAction = Callable[[Transcript], Awaitable[None] | None]
+
+
+def render_transcript_revision_row(
+    *,
+    transcript: Transcript,
+    initially_expanded: bool = False,
+    classes: str = "w-full",
+    on_delete: TranscriptAction | 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)}"
+
+    expansion = ui.expansion(value=initially_expanded, group="group").classes(
+        f"{classes} rounded-borders bg-blue-grey-10"
+    )
+
+    with expansion, ui.column().classes("w-full q-gutter-y-sm q-pa-sm"):
+        with expansion.add_slot("header"), ui.row().classes("w-full items-start justify-between q-gutter-md"):
+            with ui.column().classes("q-gutter-none"):
+                ui.label(header).classes("text-subtitle1 text-weight-medium")
+                ui.label(caption).classes("text-caption text-grey-5")
+
+            if on_delete is not None:
+                with ui.dialog() as delete_dialog, ui.card().classes("q-pa-md"):
+                    ui.label("Delete this transcript revision?").classes("text-body1")
+                    with ui.row().classes("w-full justify-end q-gutter-sm"):
+                        ui.button("Cancel", on_click=lambda: delete_dialog.submit(False)).props("flat")
+                        ui.button("Delete", on_click=lambda: delete_dialog.submit(True)).props(
+                            'unelevated color="negative"'
+                        )
+
+                async def delete_current_transcript() -> None:
+                    delete_dialog.open()
+                    confirmed = await delete_dialog
+                    if not confirmed:
+                        return
+
+                    maybe_awaitable = on_delete(transcript)
+                    if isinstance(maybe_awaitable, Awaitable):
+                        await maybe_awaitable
+
+                with ui.column(align_items="center").classes("self-center q-gutter-none"):
+                    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))
+
+        if transcript.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")
+
+    return expansion
+
+
+def _format_created_at(value: datetime) -> str:
+    """Return a compact UTC-like timestamp for row captions."""
+    return value.strftime("%Y-%m-%d %H:%M:%S %Z")
+
+
+def _metadata_row(*, label: str, value: str) -> None:
+    with ui.row().classes("w-md items-start justify-between q-gutter-x-md"):
+        ui.label(label).classes("text-caption text-grey-5 text-uppercase")
+        ui.label(value).classes("text-body2 text-right text-grey-1 break-all")

src/transcription/ui/pages/jobs_page.py

@@ -4,101 +4,96 @@ from __future__ import annotations
 
 from uuid import UUID
 
+from fastapi import Request
 from nicegui import ui
-from sqlalchemy.orm import selectinload
-from sqlmodel import desc
-from sqlmodel import select
 
-from transcription.db import get_session
-from transcription.models import Document
-from transcription.models import Job
-from transcription.models import Transcript
+from transcription.app_state import resolve_session_factory
+from transcription.models import JobStatus
+from transcription.services.jobs import JobService
+from transcription.services.transcription import TranscriptionService
 from transcription.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.error_presenter import show_error
-from transcription.ui.components.error_presenter import summarize_error
-from transcription.ui.components.job_detail import render_job_detail
-from transcription.ui.components.table.jobs import JobTableRow
 from transcription.ui.components.table.jobs import render_jobs_table
 
-
-async def fetch_job_rows() -> list[JobTableRow]:
-    """Return jobs for display in most-recent-first order."""
-    async with get_session() as session:
-        jobs = (
-            await session.exec(
-                select(Job)
-                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
-                .order_by(desc(Job.created_at))
-            )
-        ).all()
-        return [
-            JobTableRow(
-                id=job.id,
-                status=job.status.value,
-                filename=job.filename,
-                retry_count=job.retry_count,
-                created_at=job.created_at.isoformat(),
-                updated_at=job.updated_at.isoformat(),
-            )
-            for job in jobs
-        ]
-
-
-async def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]:
-    """Return job, document, and transcript for detail view."""
-    async with get_session() as session:
-        job = await session.get(Job, job_id)
-        if job is None:
-            return None, None, None
-        document = await session.get(Document, job.document_id)
-        transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job.id))).first()
-        return job, document, transcript
+from ..components.document_panzoom import render_document_panzoom
+from ..components.table.jobs import JobTableRow
+from ..components.transcript import render_transcript_revision_row
 
 
 def register_page() -> None:
     """Register jobs list and detail routes."""
 
     @ui.page("/jobs")
-    async def jobs_page() -> None:
+    async def jobs_page(request: Request) -> None:
+        session_factory = resolve_session_factory(request.app.state)
+        jobs_service = JobService(session_factory=session_factory)
         render_navigation_header(current_path="/jobs")
-        ui.label("Transcription Jobs")
-        status = ui.label("Ready")
 
         @ui.refreshable
         async def render_table() -> None:
-            jobs = await fetch_job_rows()
+            jobs = [
+                JobTableRow(
+                    id=job.id,
+                    status=job.status.value,
+                    filename=job.filename,
+                    retry_count=job.retry_count,
+                    created_at=job.created_at.isoformat(),
+                    updated_at=job.updated_at.isoformat(),
+                )
+                for job in await jobs_service.list_jobs()
+            ]
             render_jobs_table(jobs)
 
-        async def refresh() -> None:
-            status.text = "Refreshing..."
-            try:
-                await render_table.refresh()
-                status.text = "Refreshed"
-            except Exception as exc:  # noqa: BLE001
-                status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}"
-                show_error(exc, title="Jobs refresh failed", operation="jobs.refresh")
-
-        ui.button("Refresh", on_click=refresh)
+        ui.button("Refresh", on_click=render_table.refresh, icon="refresh")
         await render_table()
-        ui.link("Back to upload", "/upload")
 
     @ui.page("/jobs/{job_id}")
-    async def job_detail_page(job_id: str) -> None:
+    async def job_detail_page(job_id: str, request: Request) -> None:
+        session_factory = resolve_session_factory(request.app.state)
+        jobs_service = JobService(session_factory=session_factory)
+        transcription_service = TranscriptionService(session_factory=session_factory)
         render_navigation_header(current_path="/jobs")
-        ui.label("Job Detail")
-        try:
-            parsed_id = UUID(job_id)
-        except ValueError:
-            ui.label("Invalid job id")
-            ui.link("Back to jobs", "/jobs")
-            return
 
-        job, document, transcript = await fetch_job_detail(parsed_id)
-        if job is None:
-            ui.label("Job not found")
-            ui.link("Back to jobs", "/jobs")
-            return
+        job = await jobs_service.read_job(job_id=UUID(job_id))
 
-        render_job_detail(job=job, document=document, transcript=transcript)
+        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)
+            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)
+                with ui.row().classes("w-full items-center justify-between"):
+                    ui.label(f"{job.id}").classes("text-h6 text-weight-bold")
+                    match job.status:
+                        case JobStatus.TRANSCRIBED:
+                            ui.chip(job.status.value.upper(), color="green", text_color="white").props("outline")
+                        case _:
+                            ui.label(f"{job.status.value}").classes("text-subtitle1 text-weight-medium")
 
-        ui.link("Back to jobs", "/jobs")
+                async def delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
+                    try:
+                        transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
+                        await transcription_service.delete_transcript(transcript)
+                    except Exception as exc:  # noqa: BLE001
+                        show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
+                        return
+
+                    ui.notify(f"Deleted revision {revision}", type="positive")
+                    await render_transcript_list.refresh()
+
+                @ui.refreshable
+                async def render_transcript_list() -> 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,
+                                )
+                            ),
+                        )
+
+                await render_transcript_list()

tests/ui/conftest.py

@@ -0,0 +1,118 @@
+"""Shared fixtures for UI integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlmodel import delete
+
+from transcription.app import create_app
+from transcription.config import Settings
+from transcription.config import _settings
+from transcription.db import create_all
+from transcription.db import get_session
+from transcription.db import initialize_database_runtime
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import JobStatus
+from transcription.models import Transcript
+
+TranscriptSeed = tuple[int, str | None, str | None]
+
+
+@pytest.fixture(scope="session")
+def app_client(tmp_path_factory: pytest.TempPathFactory) -> tuple[FastAPI, TestClient]:
+    """Provide a real application and test client backed by in-memory SQLite."""
+    tmp_path = tmp_path_factory.mktemp("ui")
+    settings = Settings(
+        openrouter_api_key="test-key",
+        database_url="sqlite:///:memory:",
+        environment="test",
+        bootstrap_schema_on_startup=True,
+        upload_dir=tmp_path / "uploads",
+        prompt_dir=tmp_path / "prompts",
+    )
+    _settings.set(settings)
+
+    app = create_app()
+    app.state.runtime = initialize_database_runtime(settings=settings)
+    asyncio.run(create_all(engine=app.state.runtime.engine))
+    with TestClient(app) as client:
+        yield app, client
+
+
+@pytest.fixture(autouse=True)
+def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None:
+    """Reset UI-facing tables before each test for isolation."""
+    app, _ = app_client
+
+    async def _clear() -> None:
+        async with get_session(session_factory=app.state.runtime.session_factory) as session:
+            await session.exec(delete(Transcript))
+            await session.exec(delete(Job))
+            await session.exec(delete(Document))
+            await session.commit()
+
+    asyncio.run(_clear())
+
+
+@pytest.fixture
+def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
+    """Return a helper for inserting a document/job/transcript trio."""
+    app, _ = app_client
+    fixtures_dir = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid"
+
+    def _seed(
+        *,
+        filename: str = "sample.pdf",
+        status: JobStatus = JobStatus.TRANSCRIBED,
+        transcript_text: str | None = "Sample transcript text",
+        error_detail: str | None = None,
+        transcript_revisions: list[TranscriptSeed] | None = None,
+        source_file: Path | None = None,
+    ) -> UUID:
+        async def _insert() -> UUID:
+            async with get_session(session_factory=app.state.runtime.session_factory) as session:
+                stored_path = app.state.settings.upload_dir / filename
+                stored_path.parent.mkdir(parents=True, exist_ok=True)
+                source_path = source_file or fixtures_dir / "small_png.png"
+                stored_path.write_bytes(source_path.read_bytes())
+
+                document = Document(filename=filename, file_path=str(stored_path))
+                session.add(document)
+                await session.flush()
+
+                job = Job(document_id=document.id, status=status, retry_count=0)
+                session.add(job)
+                await session.flush()
+
+                revisions = transcript_revisions
+                if revisions is None and (transcript_text is not None or error_detail is not None):
+                    revisions = [(0, transcript_text, error_detail)]
+
+                if revisions is not None:
+                    for revision, revision_text, revision_error in revisions:
+                        session.add(
+                            Transcript(
+                                job_id=job.id,
+                                revision=revision,
+                                provider="openrouter",
+                                model="google/gemini-2.5-flash",
+                                prompt_name="transcribe_document",
+                                text=revision_text,
+                                error_detail=revision_error,
+                            )
+                        )
+
+                await session.commit()
+                return job.id
+
+        return asyncio.run(_insert())
+
+    return _seed

tests/ui/test_jobs_page.py

@@ -1,53 +1,77 @@
 """Tests for the jobs page route."""
 
-from uuid import UUID
+from pathlib import Path
+from uuid import uuid4
 
 import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
 
-from transcription.ui import register_pages
-from transcription.ui.pages import jobs_page
-from transcription.ui.pages.jobs_page import JobTableRow
-
-
-@pytest.fixture
-def client(monkeypatch):
-    """Provide a minimal app client with jobs data patched for rendering."""
-
-    async def _fetch_jobs_stub():
-        return [
-            JobTableRow(
-                id=UUID("00000000-0000-0000-0000-000000000001"),
-                status="queued",
-                filename="sample.pdf",
-                retry_count=2,
-                created_at="2026-01-01T12:00:00+00:00",
-                updated_at="2026-01-01T12:01:00+00:00",
-            )
-        ]
-
-    monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_stub)
-
-    app = FastAPI()
-    register_pages(app)
-    with TestClient(app) as test_client:
-        yield test_client
+from transcription.models import JobStatus
 
 
 @pytest.mark.integration
 class TestPageRendering:
-    """Verify the jobs page is available and includes the main controls."""
+    """Verify jobs routes render correctly with real app wiring."""
 
-    def test_jobs_page_renders_expected_controls(self, client, monkeypatch):
-        """GET /ui/jobs returns the page shell and jobs controls."""
-        response = client.get("/ui/jobs")
-
-        async def _fetch_jobs_empty():
-            return []
-
-        monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_empty)
+    def test_jobs_page_renders_empty_state(self, app_client):
+        """GET /ui/jobs renders the page and empty-state text when no jobs exist."""
+        _, client = app_client
         response = client.get("/ui/jobs")
 
         assert response.status_code == 200
+        assert "Transcription Jobs" in response.text
         assert "No jobs yet." in response.text
+
+    def test_jobs_page_lists_seeded_jobs(self, app_client, seed_job):
+        """GET /ui/jobs lists seeded jobs from the in-memory database."""
+        _, client = app_client
+        seed_job(filename="sample.pdf", status=JobStatus.TRANSCRIBED, transcript_text="done")
+
+        response = client.get("/ui/jobs")
+
+        assert response.status_code == 200
+        assert "sample.pdf" in response.text
+        assert "transcribed" in response.text
+
+    def test_job_detail_page_renders_seeded_job(self, app_client, seed_job):
+        """GET /ui/jobs/{job_id} renders detail content for a real seeded job."""
+        _, client = app_client
+        fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid" / "single_page_pdf.pdf"
+        job_id = seed_job(
+            filename="detail.pdf",
+            status=JobStatus.TRANSCRIBED,
+            transcript_revisions=[
+                (0, None, "first attempt failed"),
+                (1, "hello", None),
+            ],
+            source_file=fixture_path,
+        )
+
+        response = client.get(f"/ui/jobs/{job_id}")
+
+        assert response.status_code == 200
+        assert "Job Detail" in response.text
+        assert "Job overview" in response.text
+        assert "detail.pdf" in response.text
+        assert "Transcripts" in response.text
+        assert "Revision" in response.text
+        assert "first attempt failed" in response.text
+        assert "hello" in response.text
+        assert "Document preview" in response.text
+        assert "/uploads/detail.pdf" in response.text
+
+    def test_job_detail_page_rejects_invalid_id(self, app_client):
+        """GET /ui/jobs/{job_id} shows validation feedback for malformed IDs."""
+        _, client = app_client
+        response = client.get("/ui/jobs/not-a-uuid")
+
+        assert response.status_code == 200
+        assert "Invalid job id" in response.text
+
+    def test_job_detail_page_handles_missing_job(self, app_client):
+        """GET /ui/jobs/{job_id} shows not-found state for unknown IDs."""
+        _, client = app_client
+        missing_id = uuid4()
+        response = client.get(f"/ui/jobs/{missing_id}")
+
+        assert response.status_code == 200
+        assert "Job not found" in response.text

tests/ui/test_pages_registration.py

@@ -1,39 +1,18 @@
 """Tests for UI page registration wiring."""
 
 import pytest
-from fastapi import FastAPI
-
-from transcription.ui import register_pages
 
 
 @pytest.mark.integration
 class TestPageRegistration:
-    """Verify page registration and route wiring."""
+    """Verify page registration and mounted UI routes."""
 
-    def test_register_pages_wires_upload_jobs_and_mount(self, monkeypatch):
-        """register_pages registers pages and mounts NiceGUI at /ui."""
-        calls: list[str] = []
+    def test_ui_mount_serves_registered_pages(self, app_client):
+        """Mounted UI routes respond successfully when the full app is created."""
+        _, client = app_client
 
-        def _record_upload() -> None:
-            calls.append("upload")
+        upload_response = client.get("/ui/upload")
+        jobs_response = client.get("/ui/jobs")
 
-        def _record_jobs() -> None:
-            calls.append("jobs")
-
-        def _record_run_with(
-            _app: FastAPI,
-            *,
-            mount_path: str,
-            show_welcome_message: bool,
-            dark: bool,
-        ) -> None:
-            calls.append(f"run_with:{mount_path}:{show_welcome_message}:{dark}")
-
-        monkeypatch.setattr("transcription.ui.register_upload_page", _record_upload)
-        monkeypatch.setattr("transcription.ui.register_jobs_page", _record_jobs)
-        monkeypatch.setattr("transcription.ui.ui.run_with", _record_run_with)
-
-        app = FastAPI()
-        register_pages(app)
-
-        assert calls == ["upload", "jobs", "run_with:/ui:False:True"]
+        assert upload_response.status_code == 200
+        assert jobs_response.status_code == 200

tests/ui/test_upload_page.py

@@ -1,55 +1,35 @@
-"""Tests for the upload page route."""
-
-from pathlib import Path
+"""Tests for upload and entry-point routes."""
 
 import pytest
-from fastapi.testclient import TestClient
-
-from transcription.app import create_app
-from transcription.config import Settings
-from transcription.config import _settings
-
-
-@pytest.fixture
-def client(tmp_path: Path):
-    """Provide a real app client backed by in-memory SQLite."""
-    settings = Settings(
-        openrouter_api_key="test-key",
-        database_url="sqlite:///:memory:",
-        environment="test",
-        upload_dir=tmp_path / "uploads",
-        prompt_dir=tmp_path / "prompts",
-    )
-    _settings.set(settings)
-
-    app = create_app()
-    with TestClient(app) as test_client:
-        yield test_client
 
 
 @pytest.mark.integration
 class TestPageRendering:
-    """Verify the upload page is available and includes the main controls."""
+    """Verify upload-related routes return working pages."""
 
-    def test_root_redirects_to_ui(self, client):
+    def test_root_redirects_to_ui(self, app_client):
         """GET / redirects to the UI mount point."""
+        _, client = app_client
         response = client.get("/", follow_redirects=False)
 
         assert response.status_code == 307
         assert response.headers["location"] == "/ui"
 
-    def test_ui_redirects_to_upload(self, client):
+    def test_ui_redirects_to_upload(self, app_client):
         """GET /ui redirects to the upload page."""
+        _, client = app_client
         response = client.get("/ui", follow_redirects=False)
 
         assert response.status_code == 307
         assert response.headers["location"] == "/ui/upload"
 
-    def test_upload_page_renders_expected_controls(self, client):
+    def test_upload_page_renders_expected_controls(self, app_client):
         """GET /ui/upload returns the page shell and upload controls."""
+        _, client = app_client
         response = client.get("/ui/upload")
 
         assert response.status_code == 200
         assert "Upload Document" in response.text
         assert "Select document file" in response.text
+        assert "Upload" in response.text
         assert "Jobs" in response.text