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

README.md

@@ -10,7 +10,6 @@ The app lets you upload a document image/PDF, queues a background transcription
 - Persist document + job records in SQLite
 - Process jobs in a background worker (`queued -> processing -> transcribed/failed`)
 - Store transcript text (or failure detail)
-- Track transcript revisions (AI-generated and manual updates)
 - Show status and results in the NiceGUI interface
 
 ## Quick start
@@ -62,11 +61,8 @@ uv run uvicorn transcription.app:create_app --factory --reload
 
 - **Job detail page** (`/ui/jobs/{job_id}`)
   - Shows job metadata and status.
-  - Shows transcript metadata, including provider and model.
+  - Displays transcript text when successful.
-  - Shows a version table with `Created` and `Version`.
+  - Displays failure detail when transcription fails.
-  - Displays latest version text in an editable textbox.
-  - **Update** creates a new transcript version.
-  - Displays failure detail for failed revisions.
 
 ## Prompt artifacts
 

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,16 +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 "job" in table_names:
-    if "retry_count" not in columns:
+        job_columns = {column["name"] for column in inspector.get_columns("job")}
-        connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
+        if "retry_count" not in job_columns:
-        logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
+            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"))
+            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")
+            logger.warning("Applied SQLite compatibility schema patch table=transcript column=model default=unknown")

src/transcription/models.py

@@ -1,15 +1,12 @@
 """SQLModel domain models for the transcription system.
 
-Core models capture the MVP lifecycle:
+Three models capture the MVP lifecycle:
-Document (1) -> (many) Job
+    Document  ->  one-to-many  ->  Job  ->  one-to-many  ->  Transcript
-Job (1) -> (1) Transcript
-Job (1) -> (many) TranscriptRevision
 """
 
 from datetime import UTC
 from datetime import datetime
 from enum import StrEnum
-from typing import Optional
 from uuid import UUID
 from uuid import uuid4
 
@@ -50,8 +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")
-    transcript_revisions: list["TranscriptRevision"] = Relationship(back_populates="job")
 
     @property
     def filename(self) -> str:
@@ -63,39 +59,23 @@ 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)
+    job_id: UUID = Field(foreign_key="job.id")
-    """ID for the associated job. There's a 1-1 relationship bewteen transcripts and jobs."""
+    """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."""
-    model: str | None = None
-    """Provider model that generated the original AI transcript."""
     text: str | None = None
     """The transcribed text. This may be None if the job failed or is still in progress."""
     error_detail: str | None = None
     """Details of any error that occurred during transcription."""
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
-    # --- relationships ---
+    __table_args__ = (UniqueConstraint("job_id", "revision", name="uq_transcript_job_revision"),)
-    job: Job = Relationship(back_populates="transcript")
-
-
-class TranscriptRevision(SQLModel, table=True):
-    """Version history entries for a transcription job."""
-
-    __table_args__ = (UniqueConstraint("job_id", "version_number", name="uq_transcript_revision_job_version"),)
-
-    id: UUID = Field(default_factory=uuid4, primary_key=True)
-    job_id: UUID = Field(foreign_key="job.id", index=True)
-    version_number: int = Field(ge=1)
-    provider: str
-    prompt_name: str
-    model: str | None = None
-    source: str = Field(default="ai")
-    text: str | None = None
-    error_detail: str | None = None
-    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     # --- relationships ---
-    job: Job = Relationship(back_populates="transcript_revisions")
+    job: Job = Relationship(back_populates="transcripts")

src/transcription/providers/base.py

@@ -28,10 +28,11 @@ 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,

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
@@ -18,7 +20,6 @@ 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 TranscriptRevision
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
 from transcription.providers import ProviderResponseError
@@ -64,18 +65,6 @@ class TranscriptionService(ServiceBase):
             await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
             return transcript
 
-    async def create_transcript_revision(
-        self,
-        transcript_revision: TranscriptRevision,
-        *,
-        session: AsyncSession | None = None,
-    ) -> TranscriptRevision:
-        """Create a new transcript revision in the database."""
-        async with self._session_scope(session) as _session:
-            _session.add(transcript_revision)
-            await self._finalize(session=_session, caller_session=session, refresh=(transcript_revision,))
-            return transcript_revision
-
     async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Transcript:
         """Read an existing transcript from the database."""
         async with self._session_scope(session) as _session:
@@ -93,27 +82,6 @@ class TranscriptionService(ServiceBase):
             )
         return transcript
 
-    async def read_transcript_revision(
-        self,
-        transcript_revision_id: UUID,
-        *,
-        session: AsyncSession | None = None,
-    ) -> TranscriptRevision:
-        """Read an existing transcript revision from the database."""
-        async with self._session_scope(session) as _session:
-            transcript_revision = await _session.get(
-                TranscriptRevision,
-                transcript_revision_id,
-                options=(selectinload(TranscriptRevision.job),),  # pyright: ignore[reportArgumentType]
-            )
-        if transcript_revision is None:
-            raise TranscriptionNotFoundError(
-                f"Transcript revision with id {transcript_revision_id} not found",
-                category=ErrorCategory.NOT_FOUND,
-                suggestion="Verify the transcript revision id and retry.",
-            )
-        return transcript_revision
-
     async def update_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> Transcript:
         """Update an existing transcript in the database."""
         async with self._session_scope(session) as _session:
@@ -121,35 +89,12 @@ class TranscriptionService(ServiceBase):
             await self._finalize(session=_session, caller_session=session, refresh=(merged,))
             return merged
 
-    async def update_transcript_revision(
-        self,
-        transcript_revision: TranscriptRevision,
-        *,
-        session: AsyncSession | None = None,
-    ) -> TranscriptRevision:
-        """Update an existing transcript revision in the database."""
-        async with self._session_scope(session) as _session:
-            merged = await _session.merge(transcript_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 with self._session_scope(session) as _session:
             await _session.delete(transcript)
             await self._finalize(session=_session, caller_session=session)
 
-    async def delete_transcript_revision(
-        self,
-        transcript_revision: TranscriptRevision,
-        *,
-        session: AsyncSession | None = None,
-    ) -> None:
-        """Delete a transcript revision from the database."""
-        async with self._session_scope(session) as _session:
-            await _session.delete(transcript_revision)
-            await self._finalize(session=_session, caller_session=session)
-
     async def transcribe_document(
         self,
         image_path: str | Path,
@@ -165,95 +110,89 @@ 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,
-        prompt_name: str = DEFAULT_PROMPT_FILE,
         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()
+            rev_query = select(func.max(Transcript.revision)).where(Transcript.job_id == job_id)
-            if transcript is None:
+            rev_result = await _session.exec(rev_query)
-                transcript = Transcript(
+            max_revision = -1 if (rev := rev_result.one_or_none()) is None else rev
-                    job_id=job_id,
+            next_revision = max_revision + 1
-                    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
-            if model is not None:
-                transcript.model = model
 
+            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 list_transcript_revisions_by_job(
+    async def read_latest_transcript_by_job(
         self,
-        *,
         job_id: UUID,
+        *,
         session: AsyncSession | None = None,
-    ) -> list[TranscriptRevision]:
+    ) -> Transcript | None:
-        """Return transcript revisions for a job ordered by version number."""
+        """Read the latest transcript revision for a job id."""
         async with self._session_scope(session) as _session:
-            revisions = (
+            query = _transcript_job_query(job_id=job_id).limit(1)
-                await _session.exec(
+            result = await _session.exec(query)
-                    select(TranscriptRevision)
+            return result.one_or_none()
-                    .where(TranscriptRevision.job_id == job_id)
-                    .order_by(TranscriptRevision.version_number)
-                )
-            ).all()
-            return list(revisions)
 
-    async def append_transcript_revision(
+    async def list_transcripts_by_job(
         self,
-        *,
         job_id: UUID,
-        text: str | None,
+        *,
-        error_detail: str | None,
-        provider: str,
-        prompt_name: str,
-        model: str | None,
-        source: str,
         session: AsyncSession | None = None,
-    ) -> TranscriptRevision:
+    ) -> Sequence[Transcript]:
-        """Append a new transcript revision and allocate the next version number."""
+        """List transcript revisions for a job id in ascending revision order."""
         async with self._session_scope(session) as _session:
-            latest_version = (
+            query = _transcript_job_query(job_id=job_id)
-                await _session.exec(
+            result = await _session.exec(query)
-                    select(TranscriptRevision.version_number)
+            return result.all()
-                    .where(TranscriptRevision.job_id == job_id)
-                    .order_by(TranscriptRevision.version_number.desc())
-                    .limit(1)
-                )
-            ).first()
-            next_version = 1 if latest_version is None else latest_version + 1
 
-            revision = TranscriptRevision(
+
-                job_id=job_id,
+def _transcript_job_query(job_id: UUID):
-                version_number=next_version,
+    return (
-                provider=provider,
+        select(Transcript)
-                prompt_name=prompt_name,
+        .where(Transcript.job_id == job_id)
-                model=model,
+        .options(selectinload(Transcript.job))  # pyright: ignore[reportArgumentType]
-                source=source,
+        .order_by(Transcript.revision)  # pyright: ignore[reportArgumentType]
-                text=text,
+    )  # fmt: skip
-                error_detail=error_detail,
+
-            )
+
-            _session.add(revision)
+def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str:
-            await self._finalize(session=_session, caller_session=session, refresh=(revision,))
+    provider_model = getattr(provider, "model", None)
-            return revision
+    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(

src/transcription/services/workflows.py

@@ -121,24 +121,13 @@ async def _finalize_transcribed(
     """Transaction B: transcript + TRANSCRIBED in one commit."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            prompt_name = result.prompt_name or DEFAULT_PROMPT_FILE
+            await services.transcriptions.create_transcript_for_job(
-            await services.transcriptions.upsert_transcript_by_job(
                 job_id=job.id,
                 text=result.text,
                 error_detail=None,
                 provider=result.provider,
-                prompt_name=prompt_name,
                 model=result.model,
-                session=local_session,
+                prompt_name=result.prompt_name,
-            )
-            await services.transcriptions.append_transcript_revision(
-                job_id=job.id,
-                text=result.text,
-                error_detail=None,
-                provider=result.provider,
-                prompt_name=prompt_name,
-                model=result.model,
-                source="ai",
                 session=local_session,
             )
             updated_job = await services.jobs.mark_job_status(
@@ -149,24 +138,13 @@ async def _finalize_transcribed(
             await local_session.commit()
             return updated_job
 
-    prompt_name = result.prompt_name or DEFAULT_PROMPT_FILE
+    await services.transcriptions.create_transcript_for_job(
-    await services.transcriptions.upsert_transcript_by_job(
         job_id=job.id,
         text=result.text,
         error_detail=None,
         provider=result.provider,
-        prompt_name=prompt_name,
         model=result.model,
-        session=session,
+        prompt_name=result.prompt_name,
-    )
-    await services.transcriptions.append_transcript_revision(
-        job_id=job.id,
-        text=result.text,
-        error_detail=None,
-        provider=result.provider,
-        prompt_name=prompt_name,
-        model=result.model,
-        source="ai",
         session=session,
     )
     updated_job = await services.jobs.mark_job_status(
@@ -189,24 +167,11 @@ 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:
-            provider_name = services.transcriptions.settings.provider.value
+            await services.transcriptions.create_transcript_for_job(
-            await services.transcriptions.upsert_transcript_by_job(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
-                provider=provider_name,
                 prompt_name=DEFAULT_PROMPT_FILE,
-                model=services.transcriptions.settings.provider_model,
-                session=local_session,
-            )
-            await services.transcriptions.append_transcript_revision(
-                job_id=job.id,
-                text=None,
-                error_detail=format_error_detail(error),
-                provider=provider_name,
-                prompt_name=DEFAULT_PROMPT_FILE,
-                model=services.transcriptions.settings.provider_model,
-                source="ai",
                 session=local_session,
             )
             updated_job = await services.jobs.update_job_state(
@@ -217,24 +182,11 @@ async def _finalize_retry(
             )
             await local_session.commit()
     else:
-        provider_name = services.transcriptions.settings.provider.value
+        await services.transcriptions.create_transcript_for_job(
-        await services.transcriptions.upsert_transcript_by_job(
             job_id=job.id,
             text=None,
             error_detail=format_error_detail(error),
-            provider=provider_name,
             prompt_name=DEFAULT_PROMPT_FILE,
-            model=services.transcriptions.settings.provider_model,
-            session=session,
-        )
-        await services.transcriptions.append_transcript_revision(
-            job_id=job.id,
-            text=None,
-            error_detail=format_error_detail(error),
-            provider=provider_name,
-            prompt_name=DEFAULT_PROMPT_FILE,
-            model=services.transcriptions.settings.provider_model,
-            source="ai",
             session=session,
         )
         updated_job = await services.jobs.update_job_state(
@@ -260,24 +212,11 @@ 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:
-            provider_name = services.transcriptions.settings.provider.value
+            await services.transcriptions.create_transcript_for_job(
-            await services.transcriptions.upsert_transcript_by_job(
                 job_id=job.id,
                 text=None,
                 error_detail=format_error_detail(error),
-                provider=provider_name,
                 prompt_name=DEFAULT_PROMPT_FILE,
-                model=services.transcriptions.settings.provider_model,
-                session=local_session,
-            )
-            await services.transcriptions.append_transcript_revision(
-                job_id=job.id,
-                text=None,
-                error_detail=format_error_detail(error),
-                provider=provider_name,
-                prompt_name=DEFAULT_PROMPT_FILE,
-                model=services.transcriptions.settings.provider_model,
-                source="ai",
                 session=local_session,
             )
             updated_job = await services.jobs.mark_job_status(
@@ -288,24 +227,11 @@ async def _finalize_failed(
             await local_session.commit()
             return updated_job
 
-    provider_name = services.transcriptions.settings.provider.value
+    await services.transcriptions.create_transcript_for_job(
-    await services.transcriptions.upsert_transcript_by_job(
         job_id=job.id,
         text=None,
         error_detail=format_error_detail(error),
-        provider=provider_name,
         prompt_name=DEFAULT_PROMPT_FILE,
-        model=services.transcriptions.settings.provider_model,
-        session=session,
-    )
-    await services.transcriptions.append_transcript_revision(
-        job_id=job.id,
-        text=None,
-        error_detail=format_error_detail(error),
-        provider=provider_name,
-        prompt_name=DEFAULT_PROMPT_FILE,
-        model=services.transcriptions.settings.provider_model,
-        source="ai",
         session=session,
     )
     updated_job = await services.jobs.mark_job_status(

src/transcription/ui/__init__.py

@@ -1,14 +1,45 @@
 """UI page registration exports."""
 
+from pathlib import Path
+
 from fastapi import FastAPI
+from nicegui import app as nicegui_app
 from nicegui import ui
 
 from transcription.ui.pages.jobs_page import register_page as register_jobs_page
 from transcription.ui.pages.upload_page import register_page as register_upload_page
 
+_THEME_REGISTERED_STATE_KEY = "transcription_ui_theme_registered"
+
+_THEME_COLORS: dict[str, str] = {
+    "primary": "#6f97e8",
+    "secondary": "#92b5f5",
+    "accent": "#7fc0de",
+    "dark": "#22304a",
+    "dark_page": "#1a2538",
+    "positive": "#86c8ad",
+    "negative": "#d98a9a",
+    "info": "#7ebdda",
+    "warning": "#e2c083",
+}
+
+
+def _register_global_styles(app: FastAPI) -> None:
+    if getattr(app.state, _THEME_REGISTERED_STATE_KEY, False):
+        return
+
+    nicegui_app.colors(**_THEME_COLORS)
+
+    css_path = Path(__file__).resolve().parent / "static" / "colors.css"
+    if css_path.exists():
+        ui.add_css(css_path, shared=True)
+
+    setattr(app.state, _THEME_REGISTERED_STATE_KEY, True)
+
 
 def register_pages(app: FastAPI) -> None:
     """Register all NiceGUI pages and mount them onto the FastAPI app."""
+    _register_global_styles(app)
     register_upload_page()
     register_jobs_page()
-    ui.run_with(app, mount_path="/ui", show_welcome_message=False)
+    ui.run_with(app, mount_path="/ui", show_welcome_message=False, dark=True)

src/transcription/ui/components/__init__.py

@@ -0,0 +1,7 @@
+"""Reusable UI component exports."""
+
+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_document_panzoom", "render_navigation_header"]

src/transcription/ui/components/app_shell.py

@@ -0,0 +1,59 @@
+"""Reusable app shell primitives for page-level layout."""
+
+from __future__ import annotations
+
+from nicegui import ui
+
+NAV_ITEMS: tuple[tuple[str, str, str], ...] = (
+    ("Upload", "/upload", "upload_file"),
+    ("Jobs", "/jobs", "work_history"),
+)
+
+
+def _is_active_path(*, current_path: str, item_path: str) -> bool:
+    if item_path == "/jobs":
+        return current_path == "/jobs" or current_path.startswith("/jobs/")
+    return current_path == item_path
+
+
+def _button_props(*, icon: str, is_active: bool) -> str:
+    if is_active:
+        return f"icon={icon} no-caps unelevated color=primary text-color=white"
+    return f"icon={icon} no-caps outline color=secondary text-color=secondary"
+
+
+def _button_classes(*, is_active: bool) -> str:
+    base = "w-full sm:w-auto min-h-[40px] px-3 rounded-lg text-body2 text-weight-medium"
+    if is_active:
+        return f"{base}"
+    return f"{base}"
+
+
+def _render_nav_button(*, label: str, path: str, icon: str, current_path: str) -> None:
+    is_active = _is_active_path(current_path=current_path, item_path=path)
+    button = ui.button(
+        label,
+        icon=icon,
+        on_click=lambda _=None, route=path: ui.navigate.to(route),
+    )
+    button.props(_button_props(icon=icon, is_active=is_active)).classes(_button_classes(is_active=is_active))
+
+
+def _normalize_path(current_path: str | None) -> str:
+    normalized = (current_path or "").strip()
+    if not normalized:
+        return "/upload"
+    return normalized.rstrip("/") or "/"
+
+
+def render_navigation_header(*, current_path: str | None = None) -> None:
+    """Render a shared app header with links for top-level pages."""
+    normalized_path = _normalize_path(current_path)
+
+    with (
+        ui.header(elevated=True).props("bordered").classes("bg-dark text-white q-px-sm q-py-xs"),
+        ui.row().classes("w-full items-center justify-end q-gutter-xs"),
+        ui.element("div").classes("w-full grid grid-cols-2 gap-2 sm:flex sm:justify-start sm:gap-2"),
+    ):
+        for label, path, icon in NAV_ITEMS:
+            _render_nav_button(label=label, path=path, icon=icon, current_path=normalized_path)

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,46 +2,17 @@
 
 from __future__ import annotations
 
-import base64
+import logging
-import mimetypes
+from collections.abc import Sequence
-from collections.abc import Awaitable
-from collections.abc import Callable
-from dataclasses import dataclass
-from pathlib import Path
 
 from nicegui import ui
 
 from transcription.models import Document
 from transcription.models import Job
 from transcription.models import Transcript
-from transcription.models import TranscriptRevision
+from transcription.ui.components.document_panzoom import render_document_panzoom
 
-
+logger = logging.getLogger(__name__)
-@dataclass(frozen=True)
-class RevisionDisplayRow:
-    id: str
-    created: str
-    version: str
-    text: str
-    error_detail: str | None
-
-
-def _extract_row_id(args: object) -> str | None:
-    if isinstance(args, dict):
-        if isinstance(args.get("row"), dict):
-            row_id = args["row"].get("id")
-            return str(row_id) if row_id is not None else None
-        row_id = args.get("id")
-        return str(row_id) if row_id is not None else None
-
-    if isinstance(args, list):
-        for value in args:
-            if isinstance(value, dict):
-                row_id = value.get("id")
-                if row_id is not None:
-                    return str(row_id)
-
-    return None
 
 
 def _status_chip_classes(status: str) -> str:
@@ -57,229 +28,62 @@ def _status_chip_classes(status: str) -> str:
 
 
 def _metadata_row(label: str, value: str) -> None:
-    with ui.row().classes("w-full items-start justify-between no-wrap q-gutter-x-md"):
+    with ui.row().classes("w-full items-start justify-between q-gutter-x-md"):
-        ui.label(label).classes("text-caption text-grey-7 text-uppercase")
+        ui.label(label).classes("text-caption text-grey-5 text-uppercase w-28")
-        ui.label(value).classes("text-body2 text-right")
+        ui.label(value).classes("text-body2 text-right text-grey-1 break-all")
 
 
 def _render_document_section(document: Document) -> None:
-    with ui.card().classes("w-full bg-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")
         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")
-def _document_data_url(document: Document) -> tuple[str | None, str | None]:
+        render_document_panzoom(document=document)
-    path = Path(document.file_path)
-    if not path.exists() or not path.is_file():
-        return None, "Document preview unavailable: file not found"
-
-    suffix = path.suffix.lower()
-    mime_type, _ = mimetypes.guess_type(path.name)
-    if suffix in {".tif", ".tiff"}:
-        mime_type = "image/tiff"
-    if mime_type is None:
-        return None, "Document preview unavailable: unsupported MIME type"
-
-    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
-    return f"data:{mime_type};base64,{encoded}", None
 
 
-def _render_document_preview(document: Document) -> None:
+def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
-    source, error = _document_data_url(document)
-    if error is not None or source is None:
-        ui.label(error or "Document preview unavailable").classes("text-caption text-grey-7")
-        return
-
-    suffix = Path(document.file_path).suffix.lower()
-    if suffix == ".pdf":
-        ui.html(
-            (
-                '<iframe title="Document preview" '
-                f'src="{source}" '
-                'style="width:100%;height:520px;border:1px solid #ddd;border-radius:8px;"></iframe>'
-            )
-        )
-        ui.label("Zoom controls are currently available for image files.").classes("text-caption text-grey-7 q-mt-sm")
-        return
-
-    zoom_percent = {"value": 100}
-
-    with ui.element("div").style(
-        "width:100%;height:520px;overflow:auto;border:1px solid #ddd;border-radius:8px;padding:8px;background:#fafafa;"
-    ):
-        image = ui.image(source).classes("rounded-borders").style("width:100%;max-width:none;")
-
-    zoom_label = ui.label("Zoom: 100%").classes("text-caption text-grey-7 q-mt-sm")
-
-    def _apply_zoom() -> None:
-        image.style(f"width:{zoom_percent['value']}%;max-width:none;")
-        image.update()
-        zoom_label.text = f"Zoom: {zoom_percent['value']}%"
-        zoom_label.update()
-
-    def _zoom_in() -> None:
-        zoom_percent["value"] = min(300, zoom_percent["value"] + 25)
-        _apply_zoom()
-
-    def _zoom_out() -> None:
-        zoom_percent["value"] = max(50, zoom_percent["value"] - 25)
-        _apply_zoom()
-
-    def _zoom_reset() -> None:
-        zoom_percent["value"] = 100
-        _apply_zoom()
-
-    with ui.row().classes("q-gutter-sm q-mt-xs"):
-        ui.button("-", on_click=_zoom_out)
-        ui.button("+", on_click=_zoom_in)
-        ui.button("Reset", on_click=_zoom_reset)
-
-
-def _build_display_rows(transcript: Transcript, revisions: list[TranscriptRevision]) -> list[RevisionDisplayRow]:
-    ordered = sorted(revisions, key=lambda revision: revision.version_number)
-    rows: list[RevisionDisplayRow] = []
-
-    if ordered:
-        first = ordered[0]
-        rows.append(
-            RevisionDisplayRow(
-                id="original",
-                created=first.created_at.isoformat(),
-                version="original",
-                text=first.text or "",
-                error_detail=first.error_detail,
-            )
-        )
-        for revision in ordered[1:]:
-            rows.append(
-                RevisionDisplayRow(
-                    id=str(revision.version_number),
-                    created=revision.created_at.isoformat(),
-                    version=str(revision.version_number),
-                    text=revision.text or "",
-                    error_detail=revision.error_detail,
-                )
-            )
-    else:
-        rows.append(
-            RevisionDisplayRow(
-                id="original",
-                created=transcript.created_at.isoformat(),
-                version="original",
-                text=transcript.text or "",
-                error_detail=transcript.error_detail,
-            )
-        )
-
-    return rows
-
-
-def _render_transcript_versioned_section(
-    *,
-    document: Document | None,
-    transcript: Transcript | None,
-    revisions: list[TranscriptRevision],
-    on_update: Callable[[str], Awaitable[None]] | None,
-) -> None:
     with ui.card().classes("w-full q-pa-md"):
-        ui.label("Transcript").classes("text-subtitle1 text-weight-medium")
+        ui.label("Transcripts").classes("text-subtitle1 text-weight-medium")
         ui.separator().classes("q-my-sm")
 
-        if transcript is None:
+        if not transcripts:
-            ui.label("Transcript not available yet.").classes("text-body2 text-grey-8")
+            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"):
+        for transcript in transcripts:
-            model_name = transcript.model
+            with ui.column().classes("w-full q-gutter-y-xs"):
-            if model_name is None and revisions:
+                _metadata_row("Revision", str(transcript.revision))
-                model_name = revisions[0].model
+                _metadata_row("Provider", transcript.provider)
-            _metadata_row("Provider", transcript.provider)
+                _metadata_row("Prompt", transcript.prompt_name)
-            _metadata_row("Model", model_name or "unknown")
+                _metadata_row("Created", transcript.created_at.isoformat())
-            _metadata_row("Prompt", transcript.prompt_name)
 
-        display_rows = _build_display_rows(transcript, revisions)
+            if transcript.text:
-        rows_by_id = {row.id: row for row in display_rows}
+                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-sm")
+            ui.separator().classes("q-my-md bg-blue-grey-7")
-        ui.label("Versions").classes("text-subtitle2 text-weight-medium")
-        table = ui.table(
-            columns=[
-                {"name": "created", "label": "Created", "field": "created", "align": "left"},
-                {"name": "version", "label": "Version", "field": "version", "align": "left"},
-            ],
-            rows=[
-                {
-                    "id": row.id,
-                    "created": row.created,
-                    "version": row.version,
-                }
-                for row in display_rows
-            ],
-            row_key="id",
-        ).classes("w-full")
-
-        default_selected = display_rows[-1].id
-        selected_label = ui.label(f"Selected version: {rows_by_id[default_selected].version}").classes(
-            "text-caption text-grey-7"
-        )
-
-        ui.separator().classes("q-my-sm")
-        with ui.row().classes("w-full no-wrap items-start q-gutter-md"):
-            if document is not None:
-                with ui.column().classes("w-1/2"):
-                    ui.label("Document Preview").classes("text-subtitle2 text-weight-medium")
-                    _render_document_preview(document)
-
-            with ui.column().classes("w-1/2"):
-                editor = (
-                    ui.textarea(label="Transcript text", value=rows_by_id[default_selected].text)
-                    .props("autogrow outlined")
-                    .classes("w-full")
-                )
-                error_label = ui.label("").classes("text-body2 text-red-10")
-
-        def _set_selected(version_id: str) -> None:
-            selected = rows_by_id.get(version_id)
-            if selected is None:
-                return
-            selected_label.text = f"Selected version: {selected.version}"
-            editor.value = selected.text
-            editor.update()
-            error_label.text = selected.error_detail or ""
-            error_label.update()
-
-        def _on_row_click(event) -> None:  # noqa: ANN001
-            row_id = _extract_row_id(event.args)
-            if row_id is None:
-                return
-            _set_selected(row_id)
-
-        table.on("rowClick", _on_row_click)
-        _set_selected(default_selected)
-
-        if on_update is not None:
-            ui.button("Update", on_click=lambda: on_update(editor.value or ""))
 
 
-def render_job_detail(
+def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> None:
-    *,
-    job: Job,
-    document: Document | None,
-    transcript: Transcript | None,
-    revisions: list[TranscriptRevision],
-    on_update: Callable[[str], Awaitable[None]] | None = 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))
     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"):
             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")
-                    ui.label(str(job.id)).classes("text-caption text-grey-7")
+                    ui.label(str(job.id)).classes("text-caption text-grey-5")
                 status_chip_classes = (
                     "q-px-sm q-py-xs rounded-borders "
                     "text-weight-medium text-capitalize "
@@ -287,7 +91,7 @@ def render_job_detail(
                 )
                 ui.label(status_text).classes(status_chip_classes)
 
-            ui.separator().classes("q-my-md")
+            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())
@@ -296,9 +100,4 @@ def render_job_detail(
         if document is not None:
             _render_document_section(document)
 
-        _render_transcript_versioned_section(
+        _render_transcript_section(transcripts)
-            document=document,
-            transcript=transcript,
-            revisions=revisions,
-            on_update=on_update,
-        )

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,160 +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.app_state import resolve_session_factory
-from transcription.models import Document
+from transcription.models import JobStatus
-from transcription.models import Job
+from transcription.services.jobs import JobService
-from transcription.models import Transcript
+from transcription.services.transcription import TranscriptionService
-from transcription.models import TranscriptRevision
+from transcription.ui.components.app_shell import render_navigation_header
-from transcription.services import ServiceBundle
 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
 
-
+from ..components.document_panzoom import render_document_panzoom
-async def fetch_job_rows() -> list[JobTableRow]:
+from ..components.table.jobs import JobTableRow
-    """Return jobs for display in most-recent-first order."""
+from ..components.transcript import render_transcript_revision_row
-    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, list[TranscriptRevision]]:
-    """Return job, document, transcript snapshot, and revisions 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()
-        revisions = (
-            await session.exec(
-                select(TranscriptRevision)
-                .where(TranscriptRevision.job_id == job.id)
-                .order_by(TranscriptRevision.version_number)
-            )
-        ).all()
-        return job, document, transcript, list(revisions)
 
 
 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:
-        ui.label("Transcription Jobs")
+        session_factory = resolve_session_factory(request.app.state)
-        status = ui.label("Ready")
+        jobs_service = JobService(session_factory=session_factory)
+        render_navigation_header(current_path="/jobs")
 
         @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:
+        ui.button("Refresh", on_click=render_table.refresh, icon="refresh")
-            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)
         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:
-        ui.label("Job Detail")
+        session_factory = resolve_session_factory(request.app.state)
-        content = ui.column().classes("w-full")
+        jobs_service = JobService(session_factory=session_factory)
-        try:
+        transcription_service = TranscriptionService(session_factory=session_factory)
-            parsed_id = UUID(job_id)
+        render_navigation_header(current_path="/jobs")
-        except ValueError:
-            ui.label("Invalid job id")
-            ui.link("Back to jobs", "/jobs")
-            return
 
-        async def refresh_content() -> None:
+        job = await jobs_service.read_job(job_id=UUID(job_id))
-            content.clear()
-            job, document, transcript, revisions = await fetch_job_detail(parsed_id)
-            if job is None:
-                with content:
-                    ui.label("Job not found")
-                return
 
-            services = ServiceBundle()
+        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")
 
-            async def update_transcript_text(value: str) -> None:
+                async def delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
-                try:
+                    try:
-                    update_text = value.strip()
+                        transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
-                    async with get_session() as session:
+                        await transcription_service.delete_transcript(transcript)
-                        current_transcript = (
+                    except Exception as exc:  # noqa: BLE001
-                            await session.exec(select(Transcript).where(Transcript.job_id == parsed_id))
+                        show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
-                        ).first()
+                        return
-                        provider_name = current_transcript.provider if current_transcript is not None else "openrouter"
+
-                        prompt_name = (
+                    ui.notify(f"Deleted revision {revision}", type="positive")
-                            current_transcript.prompt_name if current_transcript is not None else "transcribe_document.md"
+                    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,
+                                )
+                            ),
                         )
-                        model_name = current_transcript.model if current_transcript is not None else None
 
-                        await services.transcriptions.upsert_transcript_by_job(
+                await render_transcript_list()
-                            job_id=parsed_id,
-                            text=update_text,
-                            error_detail=None,
-                            provider=provider_name,
-                            prompt_name=prompt_name,
-                            model=model_name,
-                            session=session,
-                        )
-                        await services.transcriptions.append_transcript_revision(
-                            job_id=parsed_id,
-                            text=update_text,
-                            error_detail=None,
-                            provider=provider_name,
-                            prompt_name=prompt_name,
-                            model=model_name,
-                            source="user",
-                            session=session,
-                        )
-                        await session.commit()
-
-                    ui.notify("Transcript updated", type="positive")
-                    await refresh_content()
-                except Exception as exc:  # noqa: BLE001
-                    show_error(exc, title="Transcript update failed", operation="jobs.detail.update")
-
-            with content:
-                render_job_detail(
-                    job=job,
-                    document=document,
-                    transcript=transcript,
-                    revisions=revisions,
-                    on_update=update_transcript_text,
-                )
-
-        await refresh_content()
-
-        ui.link("Back to jobs", "/jobs")

src/transcription/ui/pages/upload_page.py

@@ -8,6 +8,7 @@ from nicegui import ui
 from transcription.app_state import resolve_session_factory
 from transcription.db import get_session
 from transcription.services.store import create_upload_job
+from transcription.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.upload import render_upload_widget
 from transcription.worker import resolve_worker_notifier
 
@@ -17,6 +18,7 @@ def register_page() -> None:
 
     @ui.page("/upload", title="Upload Document")
     def upload_page(request: Request) -> None:
+        render_navigation_header(current_path="/upload")
         session_factory = resolve_session_factory(request.app.state)
 
         async def submit_upload(filename: str, file_bytes: bytes):
@@ -29,6 +31,3 @@ def register_page() -> None:
 
         notify_worker = resolve_worker_notifier(request.app.state)
         render_upload_widget(submitter=submit_upload, notifier=notify_worker)
-
-        with ui.row():
-            ui.link("View jobs", "/jobs")

src/transcription/ui/static/colors.css

@@ -0,0 +1,30 @@
+:root {
+	/* Soft blue-night palette tokens */
+	--ctp-rosewater: #f2dde5;
+	--ctp-flamingo: #edcfd8;
+	--ctp-pink: #dcc7de;
+	--ctp-mauve: #a9bde5;
+	--ctp-red: #d98a9a;
+	--ctp-maroon: #d39aa5;
+	--ctp-peach: #d7af8c;
+	--ctp-yellow: #e2c083;
+	--ctp-green: #86c8ad;
+	--ctp-teal: #77bfbe;
+	--ctp-sky: #7ebdda;
+	--ctp-sapphire: #74aed0;
+	--ctp-blue: #92b5f5;
+	--ctp-lavender: #6f97e8;
+	--ctp-text: #d8e2f5;
+	--ctp-subtext1: #bfcae0;
+	--ctp-subtext0: #a9b6cf;
+	--ctp-overlay2: #95a3bf;
+	--ctp-overlay1: #7c8ca9;
+	--ctp-overlay0: #657490;
+	--ctp-surface2: #4d5f7c;
+	--ctp-surface1: #394a65;
+	--ctp-surface0: #2a3954;
+	--ctp-base: #1f2b42;
+	--ctp-mantle: #1a2538;
+	--ctp-crust: #141e30;
+
+}

tests/integration/test_pipeline_flow.py

@@ -6,9 +6,9 @@ import pytest
 from sqlmodel import select
 
 from transcription.config import Settings
-from transcription.models import Job, JobStatus, Transcript, TranscriptRevision
+from transcription.models import Job, JobStatus, Transcript
 from transcription.providers.base import TranscriptionResult
-from transcription.services.store import create_upload_job
+from transcription.services.upload import create_upload_job
 from transcription.worker import process_next_queued_job
 
 
@@ -16,37 +16,24 @@ from transcription.worker import process_next_queued_job
 class TestPipelineSuccessFlow:
     """Verify end-to-end success lifecycle behavior."""
 
-    @pytest.mark.asyncio
+    def test_upload_then_worker_persists_transcribed_terminal_state(self, session, tmp_path: Path, monkeypatch):
-    async def test_upload_then_worker_persists_transcribed_terminal_state(self, async_session, tmp_path: Path, monkeypatch):
         """Upload followed by worker processing persists transcript and transcribed status."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        upload_result = await create_upload_job(
+        upload_result = create_upload_job(
             filename="pipeline.jpg",
             file_bytes=b"pipeline-bytes",
-            session=async_session,
+            session=session,
             settings=settings,
         )
 
-        async def _fake_transcribe(_path: str) -> TranscriptionResult:
+        def _fake_transcribe(_path: str) -> TranscriptionResult:
-            return TranscriptionResult(
+            return TranscriptionResult(text="Pipeline transcript", provider="openrouter", model="test-model")
-                text="Pipeline transcript",
-                provider="openrouter",
-                prompt_name="transcribe_document.md",
-                model="test-model",
-            )
 
-        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _fake_transcribe)
+        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
 
-        processed = await process_next_queued_job(session=async_session)
+        processed = process_next_queued_job(session=session)
-        job = await async_session.get(Job, upload_result.job_id)
+        job = session.get(Job, upload_result.job_id)
-        transcript = (await async_session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id))).first()
+        transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
-        revisions = (
-            await async_session.exec(
-                select(TranscriptRevision)
-                .where(TranscriptRevision.job_id == upload_result.job_id)
-                .order_by(TranscriptRevision.version_number)
-            )
-        ).all()
 
         assert processed is True
         assert job is not None
@@ -54,43 +41,30 @@ class TestPipelineSuccessFlow:
         assert transcript is not None
         assert transcript.text == "Pipeline transcript"
         assert transcript.error_detail is None
-        assert transcript.model == "test-model"
-        assert len(revisions) == 1
-        assert revisions[0].version_number == 1
-        assert revisions[0].source == "ai"
-        assert revisions[0].text == "Pipeline transcript"
 
 
 @pytest.mark.integration
 class TestPipelineFailureFlow:
     """Verify end-to-end failure lifecycle behavior."""
 
-    @pytest.mark.asyncio
+    def test_upload_then_worker_persists_failed_terminal_state(self, session, tmp_path: Path, monkeypatch):
-    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."""
         settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        upload_result = await create_upload_job(
+        upload_result = create_upload_job(
             filename="pipeline.jpg",
             file_bytes=b"pipeline-bytes",
-            session=async_session,
+            session=session,
             settings=settings,
         )
 
-        async def _fake_transcribe(_path: str) -> TranscriptionResult:
+        def _fake_transcribe(_path: str) -> TranscriptionResult:
             raise RuntimeError("pipeline provider failure")
 
-        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _fake_transcribe)
+        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
 
-        processed = await process_next_queued_job(session=async_session)
+        processed = process_next_queued_job(session=session)
-        job = await async_session.get(Job, upload_result.job_id)
+        job = session.get(Job, upload_result.job_id)
-        transcript = (await async_session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id))).first()
+        transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
-        revisions = (
-            await async_session.exec(
-                select(TranscriptRevision)
-                .where(TranscriptRevision.job_id == upload_result.job_id)
-                .order_by(TranscriptRevision.version_number)
-            )
-        ).all()
 
         assert processed is True
         assert job is not None
@@ -100,8 +74,3 @@ class TestPipelineFailureFlow:
         assert "pipeline provider failure" in transcript.error_detail
         assert "[internal_unexpected_error]" in transcript.error_detail
         assert "error_id=" in transcript.error_detail
-        assert len(revisions) == 1
-        assert revisions[0].version_number == 1
-        assert revisions[0].source == "ai"
-        assert revisions[0].text is None
-        assert "pipeline provider failure" in (revisions[0].error_detail or "")

tests/test_db.py

@@ -1,71 +1,96 @@
-"""Tests for transcription.db — async schema bootstrap/runtime behavior."""
+"""Tests for transcription.db — schema bootstrap and session factory."""
 
-from sqlalchemy import inspect
+from sqlalchemy import inspect, text
-from sqlalchemy import text
+from sqlmodel import Session, SQLModel, create_engine
-import pytest
+from sqlmodel.pool import StaticPool
+
+
+def _in_memory_engine():
+    """Create a fresh in-memory SQLite engine for isolated db tests."""
+    return create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
 
 
 class TestSchemaBootstrap:
-    """Verify async create_all produces the expected table set."""
+    """Verify create_all produces the expected table set."""
 
-    @pytest.mark.asyncio
+    def test_create_all_creates_expected_tables(self):
-    async def test_create_all_creates_expected_tables(self, default_settings):
+        """After create_all(), document, job, and transcript tables exist."""
-        """After async create_all(), document/job/transcript/revision tables exist."""
+        engine = _in_memory_engine()
-        # Ensure models are imported so metadata is populated.
+        # Ensure models are imported so metadata is populated
-        from transcription.models import Document, Job, Transcript, TranscriptRevision  # noqa: F401
+        from transcription.models import Document, Job, Transcript  # noqa: F401
 
-        from transcription.db.operations import create_all
+        import transcription.db as db_module
-        from transcription.db.runtime import get_engine
 
-        engine = get_engine(settings=default_settings)
+        db_module.create_all(engine=engine)
-        await create_all(engine=engine)
-
-        async with engine.begin() as connection:
-            table_names = set(await connection.run_sync(lambda sync_conn: inspect(sync_conn).get_table_names()))
 
+        inspector = inspect(engine)
+        table_names = set(inspector.get_table_names())
         assert "document" in table_names
         assert "job" in table_names
         assert "transcript" in table_names
-        assert "transcriptrevision" in table_names
 
 
 class TestSessionFactory:
-    """Verify async get_session yields a usable AsyncSession."""
+    """Verify get_session yields and cleans up sessions."""
 
-    @pytest.mark.asyncio
+    def test_get_session_yields_session(self):
-    async def test_get_session_yields_session(self, default_settings):
+        """get_session() yields a usable Session object."""
-        """get_session() yields an AsyncSession with a live connection."""
+        engine = _in_memory_engine()
-        from transcription.db.runtime import get_session
+        SQLModel.metadata.create_all(engine)
 
-        async with get_session(settings=default_settings) as session:
+        import transcription.db as db_module
-            result = await session.exec(text("SELECT 1"))
+
-            assert result.first()[0] == 1
+        with db_module.get_session(engine=engine) as session:
+            assert isinstance(session, Session)
+
+    def test_session_is_closed_after_generator_exit(self):
+        """After the context manager exits, the session is closed."""
+        engine = _in_memory_engine()
+        SQLModel.metadata.create_all(engine)
+
+        import transcription.db as db_module
+
+        with db_module.get_session(engine=engine) as session:
+            # Session is usable inside the context
+            session.execute(text("SELECT 1"))
+            captured = session
+
+        # After exiting, the session's internal connection is released
+        # (no active transaction bound to the session)
+        assert captured._transaction is None
 
 
 class TestBootstrapPolicy:
-    """Verify startup schema bootstrap policy via Settings property."""
+    """Verify schema bootstrap policy defaults and overrides."""
 
     def test_production_defaults_to_no_bootstrap(self):
         """Production defaults to explicit non-bootstrap startup behavior."""
         from transcription.config import Settings
+        from transcription.db import should_bootstrap_schema
 
         settings = Settings(openrouter_api_key="test-key", environment="production")
-        assert settings.should_bootstrap_schema is False
+        assert should_bootstrap_schema(settings) is False
 
     def test_development_defaults_to_bootstrap(self):
         """Development defaults to schema bootstrap for local workflows."""
         from transcription.config import Settings
+        from transcription.db import should_bootstrap_schema
 
         settings = Settings(openrouter_api_key="test-key", environment="development")
-        assert settings.should_bootstrap_schema is True
+        assert should_bootstrap_schema(settings) is True
 
     def test_explicit_override_wins(self):
         """Explicit bootstrap_schema_on_startup overrides environment default."""
         from transcription.config import Settings
+        from transcription.db import should_bootstrap_schema
 
         settings = Settings(
             openrouter_api_key="test-key",
             environment="production",
             bootstrap_schema_on_startup=True,
         )
-        assert settings.should_bootstrap_schema is True
+        assert should_bootstrap_schema(settings) is True

tests/test_models.py

@@ -1,11 +1,11 @@
-"""Tests for transcription.models — Document, Job, Transcript, TranscriptRevision models."""
+"""Tests for transcription.models — Document, Job, Transcript persistence and relationships."""
 
 from uuid import UUID
 
 import pytest
 from sqlalchemy.exc import IntegrityError
 
-from transcription.models import Document, Job, JobStatus, Transcript, TranscriptRevision
+from transcription.models import Document, Job, JobStatus, Transcript
 
 
 def _make_document(**overrides) -> Document:
@@ -113,7 +113,7 @@ class TestTranscriptModel:
         """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, provider="openrouter", prompt_name="transcribe_document.md", text="Dear Sir, ...")
+        transcript = Transcript(job_id=job.id, text="Dear Sir, ...")
         session.add(transcript)
         session.commit()
         session.refresh(transcript)
@@ -127,12 +127,7 @@ class TestTranscriptModel:
         """A Transcript with text None and error_detail set persists correctly."""
         doc = _persist_document(session)
         job = _persist_job(session, doc)
-        transcript = Transcript(
+        transcript = Transcript(job_id=job.id, error_detail="Provider timeout")
-            job_id=job.id,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            error_detail="Provider timeout",
-        )
         session.add(transcript)
         session.commit()
         session.refresh(transcript)
@@ -147,101 +142,16 @@ class TestTranscriptModel:
         doc = _persist_document(session)
         job = _persist_job(session, doc)
 
-        t1 = Transcript(job_id=job.id, provider="openrouter", prompt_name="transcribe_document.md", text="First")
+        t1 = Transcript(job_id=job.id, text="First")
         session.add(t1)
         session.commit()
 
-        t2 = Transcript(job_id=job.id, provider="openrouter", prompt_name="transcribe_document.md", text="Duplicate")
+        t2 = Transcript(job_id=job.id, text="Duplicate")
         session.add(t2)
         with pytest.raises(IntegrityError):
             session.commit()
 
 
-class TestTranscriptRevisionModel:
-    """Verify TranscriptRevision persistence and version uniqueness constraints."""
-
-    def test_revision_record_persists(self, session):
-        """A TranscriptRevision with version metadata persists correctly."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        revision = TranscriptRevision(
-            job_id=job.id,
-            version_number=1,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            model="google/gemini-2.5-flash",
-            source="ai",
-            text="Initial text",
-        )
-        session.add(revision)
-        session.commit()
-        session.refresh(revision)
-
-        fetched = session.get(TranscriptRevision, revision.id)
-        assert fetched is not None
-        assert fetched.version_number == 1
-        assert fetched.text == "Initial text"
-        assert fetched.source == "ai"
-
-    def test_job_version_pair_is_unique(self, session):
-        """Duplicate version_number for same job raises integrity error."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-
-        first = TranscriptRevision(
-            job_id=job.id,
-            version_number=1,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            source="ai",
-            text="Initial",
-        )
-        duplicate = TranscriptRevision(
-            job_id=job.id,
-            version_number=1,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            source="user",
-            text="Edited",
-        )
-        session.add(first)
-        session.commit()
-
-        session.add(duplicate)
-        with pytest.raises(IntegrityError):
-            session.commit()
-
-    def test_same_version_number_allowed_for_different_jobs(self, session):
-        """Version numbers are scoped per job, not globally."""
-        doc1 = _persist_document(session)
-        job1 = _persist_job(session, doc1)
-        doc2 = _make_document(filename="letter2.jpg", file_path="/uploads/letter2.jpg")
-        session.add(doc2)
-        session.commit()
-        session.refresh(doc2)
-        job2 = _persist_job(session, doc2)
-
-        r1 = TranscriptRevision(
-            job_id=job1.id,
-            version_number=1,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            source="ai",
-            text="Job1 v1",
-        )
-        r2 = TranscriptRevision(
-            job_id=job2.id,
-            version_number=1,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            source="ai",
-            text="Job2 v1",
-        )
-        session.add(r1)
-        session.add(r2)
-        session.commit()
-
-
 class TestRelationships:
     """Verify SQLModel relationship navigation between models."""
 
@@ -259,12 +169,7 @@ class TestRelationships:
         """job.transcript returns the linked Transcript."""
         doc = _persist_document(session)
         job = _persist_job(session, doc)
-        transcript = Transcript(
+        transcript = Transcript(job_id=job.id, text="Transcribed text")
-            job_id=job.id,
-            provider="openrouter",
-            prompt_name="transcribe_document.md",
-            text="Transcribed text",
-        )
         session.add(transcript)
         session.commit()
 
@@ -272,32 +177,3 @@ class TestRelationships:
         assert job.transcript is not None
         assert isinstance(job.transcript, Transcript)
         assert job.transcript.text == "Transcribed text"
-
-    def test_job_exposes_transcript_revisions(self, session):
-        """job.transcript_revisions returns all linked revisions."""
-        doc = _persist_document(session)
-        job = _persist_job(session, doc)
-        session.add(
-            TranscriptRevision(
-                job_id=job.id,
-                version_number=1,
-                provider="openrouter",
-                prompt_name="transcribe_document.md",
-                source="ai",
-                text="v1",
-            )
-        )
-        session.add(
-            TranscriptRevision(
-                job_id=job.id,
-                version_number=2,
-                provider="openrouter",
-                prompt_name="transcribe_document.md",
-                source="user",
-                text="v2",
-            )
-        )
-        session.commit()
-
-        session.refresh(job)
-        assert len(job.transcript_revisions) == 2

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,37 +1,77 @@
 """Tests for the jobs page route."""
 
+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.models import JobStatus
-from transcription.ui.pages import jobs_page
-
-
-@pytest.fixture
-def client(monkeypatch):
-    """Provide a minimal app client with jobs data patched for rendering."""
-
-    async def _fetch_jobs_stub():
-        return []
-
-    monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_stub)
-
-    app = FastAPI()
-    register_pages(app)
-    with TestClient(app) as test_client:
-        yield test_client
 
 
 @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):
+    def test_jobs_page_renders_empty_state(self, app_client):
-        """GET /ui/jobs returns the page shell and jobs controls."""
+        """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 "Refresh" in response.text
+        assert "No jobs yet." in response.text
-        assert "Back to upload" 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,38 +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):
+    def test_ui_mount_serves_registered_pages(self, app_client):
-        """register_pages registers pages and mounts NiceGUI at /ui."""
+        """Mounted UI routes respond successfully when the full app is created."""
-        calls: list[str] = []
+        _, client = app_client
 
-        def _record_upload() -> None:
+        upload_response = client.get("/ui/upload")
-            calls.append("upload")
+        jobs_response = client.get("/ui/jobs")
 
-        def _record_jobs() -> None:
+        assert upload_response.status_code == 200
-            calls.append("jobs")
+        assert jobs_response.status_code == 200
-
-        def _record_run_with(
-            _app: FastAPI,
-            *,
-            mount_path: str,
-            show_welcome_message: bool,
-        ) -> None:
-            calls.append(f"run_with:{mount_path}:{show_welcome_message}")
-
-        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"]

tests/ui/test_upload_page.py

@@ -1,55 +1,35 @@
-"""Tests for the upload page route."""
+"""Tests for upload and entry-point routes."""
-
-from pathlib import Path
 
 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 "View jobs" in response.text
+        assert "Upload" in response.text
+        assert "Jobs" in response.text