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

.vscode/launch.json

@@ -0,0 +1,27 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Debug transcription app",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "debugpy",
+            "args": [
+                "-m",
+                "uvicorn",
+                "transcription.app:create_app",
+                "--factory",
+                "--host",
+                // "127.0.0.1",
+                "0.0.0.0",
+                "--port",
+                "8080"
+            ],
+            "justMyCode": true,
+            "console": "integratedTerminal",
+            "env": {
+                "PYTHONPATH": "${workspaceFolder}/src"
+            }
+        }
+    ]
+}

src/transcription/app.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
-import asyncio
+from contextlib import AsyncExitStack
 from contextlib import asynccontextmanager
-from contextlib import suppress
 
 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
@@ -17,49 +19,14 @@ from .db import dispose_database_runtime
 from .db import initialize_database_runtime
 from .services import ServiceBundle
 from .ui import register_pages
-from .worker import run_worker_loop
+from .worker import worker_consumer_lifespan
-
-
-def _start_worker(app: FastAPI) -> None:
-    stop_event = asyncio.Event()
-    wake_queue: asyncio.Queue[None] = asyncio.Queue()
-    worker_task = asyncio.create_task(
-        run_worker_loop(
-            session_factory=app.state.runtime.session_factory,
-            stop_event=stop_event,
-            wake_queue=wake_queue,
-            poll_interval_seconds=1.0,
-        )
-    )
-    wake_queue.put_nowait(None)
-    app.state.worker_stop_event = stop_event
-    app.state.worker_wake_queue = wake_queue
-    app.state.worker_task = worker_task
-
-
-async def _stop_worker(app: FastAPI) -> None:
-    stop_event = getattr(app.state, "worker_stop_event", None)
-    wake_queue = getattr(app.state, "worker_wake_queue", None)
-    worker_task = getattr(app.state, "worker_task", None)
-
-    if stop_event is not None:
-        stop_event.set()
-    if wake_queue is not None:
-        wake_queue.put_nowait(None)
-    if worker_task is not None:
-        try:
-            await asyncio.wait_for(worker_task, timeout=2.0)
-        except TimeoutError:
-            worker_task.cancel()
-            with suppress(asyncio.CancelledError):
-                await worker_task
 
 
 @asynccontextmanager
 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)
@@ -70,17 +37,38 @@ async def _lifespan(app: FastAPI):
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
 
-    _start_worker(app)
+    async with AsyncExitStack() as stack:
-    try:
+        stack.push_async_callback(dispose_database_runtime)
+        stop_event, worker_notifier = await stack.enter_async_context(
+            worker_consumer_lifespan(
+                session_factory=app.state.runtime.session_factory,
+                poll_interval_seconds=1.0,
+            )
+        )
+        app.state.worker_stop_event = stop_event
+        app.state.worker_notifier = worker_notifier
         yield
-    finally:
-        await _stop_worker(app)
-        await dispose_database_runtime()
 
 
 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:
+        return RedirectResponse(url="/ui", status_code=status.HTTP_307_TEMPORARY_REDIRECT)
+
+    @app.get("/ui", include_in_schema=False)
+    async def ui_redirect() -> RedirectResponse:
+        return RedirectResponse(url="/ui/upload", status_code=status.HTTP_307_TEMPORARY_REDIRECT)
+
     register_error_handlers(app)
     register_pages(app)
     app.include_router(health_router)

src/transcription/app_state.py

@@ -0,0 +1,39 @@
+"""Helpers for accessing lifespan-owned application state resources."""
+
+from __future__ import annotations
+
+from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from transcription.db.runtime import DatabaseRuntime
+from transcription.db.runtime import get_session_factory
+from transcription.worker import WorkerNotifier
+from transcription.worker import resolve_worker_notifier
+
+
+def resolve_database_runtime(state: object) -> DatabaseRuntime | None:
+    """Return database runtime from app-like state objects when available."""
+    runtime = getattr(state, "runtime", None)
+    return runtime if isinstance(runtime, DatabaseRuntime) else None
+
+
+def require_database_runtime(state: object) -> DatabaseRuntime:
+    """Return database runtime or raise when app lifespan has not initialized it."""
+    runtime = resolve_database_runtime(state)
+    if runtime is None:
+        raise RuntimeError("Database runtime is not initialized on application state")
+    return runtime
+
+
+def resolve_session_factory(state: object) -> async_sessionmaker[AsyncSession]:
+    """Return DB session factory from state when available, otherwise shared runtime."""
+    runtime = resolve_database_runtime(state)
+    if runtime is not None:
+        return runtime.session_factory
+    return get_session_factory()
+
+
+def get_worker_notifier(app: FastAPI) -> WorkerNotifier:
+    """Return app worker notifier, or a no-op fallback when unavailable."""
+    return resolve_worker_notifier(app.state)

src/transcription/db/operations.py

@@ -51,10 +51,15 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
 
     inspector = inspect(connection)
     table_names = set(inspector.get_table_names())
-    if "job" not in table_names:
-        return
 
-    columns = {column["name"] for column in inspector.get_columns("job")}
+    if "job" in table_names:
-    if "retry_count" not in columns:
+        job_columns = {column["name"] for column in inspector.get_columns("job")}
+        if "retry_count" not in job_columns:
             connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
             logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
+
+    if "transcript" in table_names:
+        transcript_columns = {column["name"] for column in inspector.get_columns("transcript")}
+        if "model" not in transcript_columns:
+            connection.execute(text("ALTER TABLE transcript ADD COLUMN model VARCHAR NOT NULL DEFAULT 'unknown'"))
+            logger.warning("Applied SQLite compatibility schema patch table=transcript column=model default=unknown")

src/transcription/models.py

@@ -1,16 +1,16 @@
 """SQLModel domain models for the transcription system.
 
 Three models capture the MVP lifecycle:
-  Document  ->  one-to-many  ->  Job  ->  one-to-one  ->  Transcript
+    Document  ->  one-to-many  ->  Job  ->  one-to-many  ->  Transcript
 """
 
 from datetime import UTC
 from datetime import datetime
 from enum import StrEnum
-from typing import Optional
 from uuid import UUID
 from uuid import uuid4
 
+from sqlalchemy import UniqueConstraint
 from sqlmodel import Field
 from sqlmodel import Relationship
 from sqlmodel import SQLModel
@@ -47,7 +47,7 @@ class Job(SQLModel, table=True):
 
     # --- relationships ---
     document: Document = Relationship(back_populates="jobs")
-    transcript: Optional["Transcript"] = Relationship(back_populates="job")
+    transcripts: list["Transcript"] = Relationship(back_populates="job")
 
     @property
     def filename(self) -> str:
@@ -59,10 +59,14 @@ class Transcript(SQLModel, table=True):
     """The output of a transcription job."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
-    job_id: UUID = Field(foreign_key="job.id", unique=True)
+    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."""
     text: str | None = None
@@ -71,5 +75,7 @@ class Transcript(SQLModel, table=True):
     """Details of any error that occurred during transcription."""
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
+    __table_args__ = (UniqueConstraint("job_id", "revision", name="uq_transcript_job_revision"),)
+
     # --- relationships ---
-    job: Job = Relationship(back_populates="transcript")
+    job: Job = Relationship(back_populates="transcripts")

src/transcription/providers/base.py

@@ -28,12 +28,14 @@ class TranscriptionResult:
     prompt_name: str
     model: str
 
-    def to_transcript(self, job_id: UUID) -> Transcript:
+    def to_transcript(self, job_id: UUID, *, revision: int = 0) -> Transcript:
         """Convert a TranscriptionResult to a Transcript model instance."""
         return Transcript(
             job_id=job_id,
+            revision=revision,
             provider=self.provider,
             prompt_name=self.prompt_name,
+            model=self.model,
             text=self.text,
         )
 

src/transcription/services/jobs.py

@@ -35,7 +35,10 @@ class JobService(ServiceBase):
         async with self._session_scope(session) as _session:
             query = (
                 select(Job)
-                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .options(
+                    selectinload(Job.document),  # pyright: ignore[reportArgumentType]
+                    selectinload(Job.transcripts),  # pyright: ignore[reportArgumentType]
+                )
                 .where(Job.id == job_id)
                 .execution_options(populate_existing=True)
             )

src/transcription/services/transcription.py

@@ -4,10 +4,12 @@ from __future__ import annotations
 
 import logging
 import mimetypes
+from collections.abc import Sequence
 from contextlib import contextmanager
 from pathlib import Path
 from uuid import UUID
 
+from sqlalchemy import func
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlalchemy.orm import selectinload
 from sqlmodel import select
@@ -108,38 +110,90 @@ class TranscriptionService(ServiceBase):
             settings=self.settings,
             provider=self.provider,
         )
-        await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session)
+        await self.create_transcript_for_job(
+            job_id=job_id,
+            text=result.text,
+            provider=result.provider,
+            model=result.model,
+            prompt_name=result.prompt_name,
+            session=session,
+        )
 
-    async def upsert_transcript_by_job(
+    async def create_transcript_for_job(
         self,
         *,
         job_id: UUID,
         text: str | None,
-        error_detail: str | None,
+        error_detail: str | None = None,
         provider: str | None = None,
+        model: str | None = None,
         prompt_name: str = DEFAULT_PROMPT_FILE,
         session: AsyncSession | None = None,
     ) -> Transcript:
-        """Create or update a transcript for a job id."""
+        """Create a new transcript revision for a job id."""
         async with self._session_scope(session) as _session:
-            transcript = (await _session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
+            rev_query = select(func.max(Transcript.revision)).where(Transcript.job_id == job_id)
-            if transcript is None:
+            rev_result = await _session.exec(rev_query)
+            max_revision = -1 if (rev := rev_result.one_or_none()) is None else rev
+            next_revision = max_revision + 1
+
             transcript = Transcript(
                 job_id=job_id,
+                revision=next_revision,
                 provider=provider or self.settings.provider.value,
+                model=model or _resolve_transcript_model(provider=self.provider, settings=self.settings),
                 prompt_name=prompt_name,
+                text=text,
+                error_detail=error_detail,
             )
-
-            transcript.text = text
-            transcript.error_detail = error_detail
-            if provider is not None:
-                transcript.provider = provider
-            transcript.prompt_name = prompt_name
-
             _session.add(transcript)
             await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
             return transcript
 
+    async def read_latest_transcript_by_job(
+        self,
+        job_id: UUID,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Transcript | None:
+        """Read the latest transcript revision for a job id."""
+        async with self._session_scope(session) as _session:
+            query = _transcript_job_query(job_id=job_id).limit(1)
+            result = await _session.exec(query)
+            return result.one_or_none()
+
+    async def list_transcripts_by_job(
+        self,
+        job_id: UUID,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Transcript]:
+        """List transcript revisions for a job id in ascending revision order."""
+        async with self._session_scope(session) as _session:
+            query = _transcript_job_query(job_id=job_id)
+            result = await _session.exec(query)
+            return result.all()
+
+
+def _transcript_job_query(job_id: UUID):
+    return (
+        select(Transcript)
+        .where(Transcript.job_id == job_id)
+        .options(selectinload(Transcript.job))  # pyright: ignore[reportArgumentType]
+        .order_by(Transcript.revision)  # pyright: ignore[reportArgumentType]
+    )  # fmt: skip
+
+
+def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str:
+    provider_model = getattr(provider, "model", None)
+    if isinstance(provider_model, str) and provider_model.strip():
+        return provider_model
+
+    if settings.provider_model and settings.provider_model.strip():
+        return settings.provider_model
+
+    return "unknown"
+
 
 async def transcribe_document_image(
     image_path: str | Path,

src/transcription/services/workflows.py

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

src/transcription/ui/__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,29 +2,102 @@
 
 from __future__ import annotations
 
+import logging
+from collections.abc import Sequence
+
 from nicegui import ui
 
 from transcription.models import Document
 from transcription.models import Job
 from transcription.models import Transcript
+from transcription.ui.components.document_panzoom import render_document_panzoom
+
+logger = logging.getLogger(__name__)
 
 
-def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None:
+def _status_chip_classes(status: str) -> str:
+    if status == "queued":
+        return "bg-blue-1 text-blue-10"
+    if status == "processing":
+        return "bg-amber-1 text-amber-10"
+    if status == "transcribed":
+        return "bg-green-1 text-green-10"
+    if status == "failed":
+        return "bg-red-1 text-red-10"
+    return "bg-grey-2 text-grey-9"
+
+
+def _metadata_row(label: str, value: str) -> None:
+    with ui.row().classes("w-full items-start justify-between q-gutter-x-md"):
+        ui.label(label).classes("text-caption text-grey-5 text-uppercase w-28")
+        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 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")
+        render_document_panzoom(document=document)
+
+
+def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
+    with ui.card().classes("w-full q-pa-md"):
+        ui.label("Transcripts").classes("text-subtitle1 text-weight-medium")
+        ui.separator().classes("q-my-sm")
+
+        if not transcripts:
+            ui.label("Transcript history is not available yet.").classes("text-body2 text-grey-3")
+            return
+
+        for transcript in transcripts:
+            with ui.column().classes("w-full q-gutter-y-xs"):
+                _metadata_row("Revision", str(transcript.revision))
+                _metadata_row("Provider", transcript.provider)
+                _metadata_row("Prompt", transcript.prompt_name)
+                _metadata_row("Created", transcript.created_at.isoformat())
+
+            if transcript.text:
+                ui.separator().classes("q-my-sm")
+                with ui.card().classes("w-fullq-pa-sm"):
+                    ui.markdown(transcript.text).classes("text-grey-1")
+            elif transcript.error_detail:
+                ui.separator().classes("q-my-sm")
+                with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
+                    ui.label("Failure detail").classes("text-caption text-uppercase")
+                    ui.label(transcript.error_detail).classes("text-body2")
+
+            ui.separator().classes("q-my-md bg-blue-grey-7")
+
+
+def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> None:
     """Render all sections for the job detail page."""
-    ui.label(f"Job ID: {job.id}")
+    logger.debug("Rendering job detail for job ID %s with %d transcripts", job.id, len(transcripts))
-    ui.label(f"Status: {job.status.value}")
+    status_text = job.status.value
-    ui.label(f"Created: {job.created_at.isoformat()}")
+    with ui.column().classes("w-full max-w-4xl q-gutter-md"):
-    ui.label(f"Updated: {job.updated_at.isoformat()}")
+        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-5")
+                status_chip_classes = (
+                    "q-px-sm q-py-xs rounded-borders "
+                    "text-weight-medium text-capitalize "
+                    f"{_status_chip_classes(status_text)}"
+                )
+                ui.label(status_text).classes(status_chip_classes)
+
+            ui.separator().classes("q-my-md bg-blue-grey-7")
+            with ui.column().classes("w-full q-gutter-y-xs"):
+                _metadata_row("Created", job.created_at.isoformat())
+                _metadata_row("Updated", job.updated_at.isoformat())
+                _metadata_row("Retries", str(job.retry_count))
 
         if document is not None:
-        ui.label(f"Filename: {document.filename}")
+            _render_document_section(document)
-        ui.label(f"File path: {document.file_path}")
 
-    if transcript is None:
+        _render_transcript_section(transcripts)
-        ui.label("Transcript not available yet.")
-    elif transcript.text:
-        ui.label("Transcript:")
-        ui.markdown(transcript.text)
-    elif transcript.error_detail:
-        ui.label("Failure detail:")
-        ui.label(transcript.error_detail)

src/transcription/ui/components/job_table.py

@@ -1,55 +0,0 @@
-"""Reusable jobs table rendering helpers."""
-
-from __future__ import annotations
-
-from collections.abc import Sequence
-from dataclasses import dataclass
-from uuid import UUID
-
-from nicegui import ui
-
-
-@dataclass(frozen=True)
-class JobTableRow:
-    """Read model consumed by the shared jobs table component."""
-
-    id: UUID
-    status: str
-    created_at: str
-    updated_at: str
-
-
-def _serialize_rows(rows: Sequence[JobTableRow]) -> list[dict[str, str]]:
-    """Convert typed rows into table-compatible dictionaries."""
-    return [
-        {
-            "id": str(row.id),
-            "status": row.status,
-            "created_at": row.created_at,
-            "updated_at": row.updated_at,
-        }
-        for row in rows
-    ]
-
-
-def render_jobs_table(rows: Sequence[JobTableRow]) -> None:
-    """Render jobs table and per-row detail links."""
-    if not rows:
-        ui.label("No jobs yet.")
-        return
-
-    serialized_rows = _serialize_rows(rows)
-    ui.table(
-        columns=[
-            {"name": "id", "label": "Job ID", "field": "id"},
-            {"name": "status", "label": "Status", "field": "status"},
-            {"name": "created_at", "label": "Created", "field": "created_at"},
-            {"name": "updated_at", "label": "Updated", "field": "updated_at"},
-        ],
-        rows=serialized_rows,
-        row_key="id",
-    ).classes("w-full")
-
-    with ui.column().classes("gap-1"):
-        for row in serialized_rows:
-            ui.link(f"Open {row['id']}", f"/jobs/{row['id']}")

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

@@ -0,0 +1,4 @@
+from .jobs import JobTableRow
+from .jobs import render_jobs_table
+
+__all__ = ["JobTableRow", "render_jobs_table"]

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

@@ -0,0 +1,73 @@
+"""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):
+        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 _bind_row_click_handler(
+    table: Any,
+    *,
+    on_row_click_id: Callable[[str], None],
+) -> None:
+    def handle_row_click(event: events.GenericEventArguments) -> None:
+        row_id = _extract_row_id(event.args)
+        if row_id is None:
+            return
+        on_row_click_id(row_id)
+
+    table.on("rowClick", handle_row_click)
+    logger.debug("Row click handler bound to table")
+
+
+def build_table(
+    rows: list[dict[str, Any]],
+    columns: list[dict[str, Any]],
+    *,
+    default_sort_by: str | None = None,
+    default_descending: bool = False,
+    classes: str = "app-table",
+    on_row_click_id: Callable[[str], None] | None = None,
+) -> Any:
+    pagination: dict[str, Any] = {"rowsPerPage": 25}
+    if default_sort_by is not None:
+        pagination["sortBy"] = default_sort_by
+        pagination["descending"] = default_descending
+
+    table = (
+        ui.table(
+            rows=rows,
+            columns=columns,
+            row_key="id",
+            pagination=pagination,
+        )
+        .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/table/jobs.py

@@ -0,0 +1,75 @@
+"""Jobs table rendering helpers."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from nicegui import ui
+
+from .common import build_table
+
+
+@dataclass(frozen=True, slots=True)
+class JobTableRow:
+    """Read model consumed by the jobs table component."""
+
+    id: UUID
+    status: str
+    filename: str
+    retry_count: int
+    created_at: str
+    updated_at: str
+
+
+def _format_timestamp(value: str) -> str:
+    """Return a friendly UTC timestamp for table display."""
+    try:
+        parsed = datetime.fromisoformat(value)
+    except ValueError:
+        return value
+    parsed = parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC)
+    return parsed.astimezone().strftime("%b %d, %I:%M %p")
+
+
+def _serialize_rows(rows: Sequence[JobTableRow]) -> list[dict[str, Any]]:
+    return [
+        {
+            "id": str(row.id),
+            "status": row.status,
+            "filename": row.filename,
+            "retry_count": row.retry_count,
+            "created_at": _format_timestamp(row.created_at),
+            "updated_at": _format_timestamp(row.updated_at),
+            "created_sort": row.created_at,
+            "updated_sort": row.updated_at,
+        }
+        for row in rows
+    ]
+
+
+def render_jobs_table(rows: Sequence[JobTableRow]) -> None:
+    """Render jobs table and open a detail page when clicking a row."""
+    if not rows:
+        ui.label("No jobs yet.")
+        return
+
+    build_table(
+        rows=_serialize_rows(rows),
+        columns=[
+            {"name": "id", "label": "Job ID", "field": "id", "sortable": True},
+            {"name": "status", "label": "Status", "field": "status", "sortable": True},
+            {"name": "filename", "label": "Filename", "field": "filename", "sortable": True},
+            {"name": "retry_count", "label": "Retries", "field": "retry_count", "sortable": True},
+            {"name": "created_at", "label": "Created", "field": "created_at", "sortable": True},
+            {"name": "updated_at", "label": "Updated", "field": "updated_at", "sortable": True},
+        ],
+        default_sort_by="created_sort",
+        default_descending=True,
+        classes="app-table w-full",
+        on_row_click_id=lambda job_id: ui.navigate.to(f"/jobs/{job_id}"),
+    )

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/components/upload.py

@@ -9,11 +9,11 @@ from nicegui import ui
 from nicegui.binding import bindable_dataclass
 from nicegui.events import UploadEventArguments
 
-from transcription.services.documents import DocumentService
+from transcription.errors import AppError
-from transcription.services.documents import UploadError
 from transcription.services.documents import UploadJobResult
 from transcription.ui.components.error_presenter import show_error
 from transcription.ui.components.error_presenter import summarize_error
+from transcription.worker import WorkerNotifier
 
 type UploadSubmitter = Callable[[str, bytes], Awaitable[UploadJobResult]]
 
@@ -26,7 +26,7 @@ class UploadWidgetState:
     message: str = ""
 
 
-def render_upload_widget(*, service: DocumentService) -> None:
+def render_upload_widget(*, submitter: UploadSubmitter, notifier: WorkerNotifier | None = None) -> None:
     """Render upload controls and common status/error handling."""
     state = UploadWidgetState()
     status_label = ui.label("Upload a document to start transcription.")
@@ -41,12 +41,14 @@ def render_upload_widget(*, service: DocumentService) -> None:
         status_label.text = "Uploading..."
         try:
             payload = await event.file.read()
-            result = await service.upload_file(filename=event.file.name, file_bytes=payload)
+            result = await submitter(event.file.name, payload)
             job_id = result.job_id
             state.message = f"Created job {job_id}" if job_id is not None else "Upload complete"
             status_label.text = state.message
+            if notifier is not None:
+                notifier.notify()
             ui.notify(state.message, type="positive")
-        except UploadError as exc:
+        except AppError as exc:
             state.message = summarize_error(exc, operation="upload.submit")
             status_label.text = f"Upload failed: {state.message}"
             show_error(exc, title="Upload failed", operation="upload.submit")

src/transcription/ui/pages/jobs_page.py

@@ -4,89 +4,96 @@ from __future__ import annotations
 
 from uuid import UUID
 
+from fastapi import Request
 from nicegui import ui
-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.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.error_presenter import show_error
-from transcription.ui.components.error_presenter import summarize_error
+from transcription.ui.components.table.jobs import render_jobs_table
-from transcription.ui.components.job_detail import render_job_detail
-from transcription.ui.components.job_table import JobTableRow
-from transcription.ui.components.job_table import render_jobs_table
 
-
+from ..components.document_panzoom import render_document_panzoom
-async def fetch_jobs() -> 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).order_by(desc(Job.created_at)))).all()
-        return [
-            JobTableRow(
-                id=job.id,
-                status=job.status.value,
-                created_at=job.created_at.isoformat(),
-                updated_at=job.updated_at.isoformat(),
-            )
-            for job in jobs
-        ]
-
-
-async def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]:
-    """Return job, document, and transcript for detail view."""
-    async with get_session() as session:
-        job = await session.get(Job, job_id)
-        if job is None:
-            return None, None, None
-        document = await session.get(Document, job.document_id)
-        transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job.id))).first()
-        return job, document, transcript
 
 
 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_jobs()
+            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", "/")
 
     @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)
+        jobs_service = JobService(session_factory=session_factory)
+        transcription_service = TranscriptionService(session_factory=session_factory)
+        render_navigation_header(current_path="/jobs")
+
+        job = await jobs_service.read_job(job_id=UUID(job_id))
+
+        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 delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
                     try:
-            parsed_id = UUID(job_id)
+                        transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
-        except ValueError:
+                        await transcription_service.delete_transcript(transcript)
-            ui.label("Invalid job id")
+                    except Exception as exc:  # noqa: BLE001
-            ui.link("Back to jobs", "/jobs")
+                        show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
                         return
 
-        job, document, transcript = await fetch_job_detail(parsed_id)
+                    ui.notify(f"Deleted revision {revision}", type="positive")
-        if job is None:
+                    await render_transcript_list.refresh()
-            ui.label("Job not found")
-            ui.link("Back to jobs", "/jobs")
-            return
 
-        render_job_detail(job=job, document=document, transcript=transcript)
+                @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,
+                                )
+                            ),
+                        )
 
-        ui.link("Back to jobs", "/jobs")
+                await render_transcript_list()

src/transcription/ui/pages/upload_page.py

@@ -2,20 +2,32 @@
 
 from __future__ import annotations
 
+from fastapi import Request
 from nicegui import ui
 
-from transcription.services.documents import DocumentService
+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
 
 
 def register_page() -> None:
     """Register the upload page route."""
 
-    service = DocumentService()
-
     @ui.page("/upload", title="Upload Document")
-    def upload_page() -> None:
+    def upload_page(request: Request) -> None:
-        render_upload_widget(service=service)
+        render_navigation_header(current_path="/upload")
+        session_factory = resolve_session_factory(request.app.state)
 
-        with ui.row():
+        async def submit_upload(filename: str, file_bytes: bytes):
-            ui.link("View jobs", "/jobs")
+            async with get_session(session_factory=session_factory) as session:
+                return await create_upload_job(
+                    filename=filename,
+                    file_bytes=file_bytes,
+                    session=session,
+                )
+
+        notify_worker = resolve_worker_notifier(request.app.state)
+        render_upload_widget(submitter=submit_upload, notifier=notify_worker)

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;
+
+}

src/transcription/worker.py

@@ -8,6 +8,7 @@ from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from contextlib import contextmanager
 from contextlib import suppress
+from typing import Protocol
 from uuid import UUID
 
 from sqlalchemy.ext.asyncio import async_sessionmaker
@@ -27,6 +28,73 @@ from .services.workflows import process_next_queued_job as process_next_queued_j
 logger = logging.getLogger(__name__)
 
 
+class WorkerNotifier(Protocol):
+    """Abstraction for signaling the worker loop about new work."""
+
+    def notify(self) -> None:
+        """Signal the worker loop that work may be available."""
+
+
+class EventWorkerNotifier:
+    """Worker notifier backed by an asyncio.Event."""
+
+    def __init__(self, wake_event: asyncio.Event):
+        self._wake_event = wake_event
+
+    def notify(self) -> None:
+        self._wake_event.set()
+
+
+class NoopWorkerNotifier:
+    """Fallback notifier used when worker signaling is unavailable."""
+
+    def notify(self) -> None:
+        return
+
+
+def resolve_worker_notifier(state: object) -> WorkerNotifier:
+    """Resolve notifier from app-like state objects with no-op fallback."""
+    notifier = getattr(state, "worker_notifier", None)
+    if isinstance(notifier, NoopWorkerNotifier):
+        return notifier
+    if notifier is None:
+        return NoopWorkerNotifier()
+    return notifier
+
+
+@asynccontextmanager
+async def worker_consumer_lifespan(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> AsyncGenerator[tuple[asyncio.Event, WorkerNotifier]]:
+    """Start and stop the worker consumer loop for app lifespan."""
+    stop_event = asyncio.Event()
+    wake_event = asyncio.Event()
+    worker_notifier: WorkerNotifier = EventWorkerNotifier(wake_event)
+    worker_task = asyncio.create_task(
+        run_worker_loop(
+            session_factory=session_factory,
+            stop_event=stop_event,
+            wake_event=wake_event,
+            poll_interval_seconds=poll_interval_seconds,
+        )
+    )
+    worker_notifier.notify()
+
+    try:
+        yield stop_event, worker_notifier
+    finally:
+        stop_event.set()
+        worker_notifier.notify()
+        try:
+            await asyncio.wait_for(worker_task, timeout=2.0)
+        except TimeoutError:
+            worker_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await worker_task
+
+
 async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Event):
     """Main worker loop that consumes jobs from the queue and processes them.
 
@@ -65,12 +133,12 @@ async def run_worker_loop(
     *,
     session_factory: async_sessionmaker[AsyncSession] | None = None,
     stop_event: asyncio.Event | None = None,
-    wake_queue: asyncio.Queue[None] | None = None,
+    wake_event: asyncio.Event | None = None,
     poll_interval_seconds: float = 1.0,
 ) -> None:
     """Run worker loop until stop_event is set.
 
-    If wake_queue is provided, queue activity wakes the loop immediately while
+    If wake_event is provided, signal activity wakes the loop immediately while
     timeout-based wakeups preserve current polling behavior.
     """
     while True:
@@ -78,15 +146,16 @@ async def run_worker_loop(
             logger.info("Worker stop event received")
             return
 
-        if wake_queue is not None:
+        if wake_event is not None:
             with suppress(TimeoutError):
-                await asyncio.wait_for(wake_queue.get(), timeout=poll_interval_seconds)
+                await asyncio.wait_for(wake_event.wait(), timeout=poll_interval_seconds)
+                wake_event.clear()
 
         processed_any = False
         while await process_next_queued_job(session_factory=session_factory):
             processed_any = True
 
-        if wake_queue is None and not processed_any:
+        if wake_event is None and not processed_any:
             await asyncio.sleep(poll_interval_seconds)
 
 

tests/conftest.py

@@ -17,6 +17,9 @@ from transcription.db.operations import create_all
 from transcription.db.runtime import dispose_database_runtime
 from transcription.db.runtime import get_engine
 from transcription.db.runtime import get_session
+from transcription.db.runtime import get_session_factory
+from transcription.services.documents import DocumentService
+from transcription.services.jobs import JobService
 
 
 @pytest.fixture
@@ -47,3 +50,22 @@ async def async_session(default_settings: Settings):
         yield async_session
 
     await dispose_database_runtime()
+
+
+@pytest.fixture
+def default_session_factory(default_settings: Settings):
+    """Provide a base fixture for tests that require database access."""
+    session_factory = get_session_factory(settings=default_settings)
+    return session_factory
+
+
+@pytest.fixture
+def job_service(default_session_factory) -> JobService:
+    """Provide a JobService instance for testing."""
+    return JobService(session_factory=default_session_factory)
+
+
+@pytest.fixture
+def document_service(default_session_factory) -> DocumentService:
+    """Provide a DocumentService instance for testing."""
+    return DocumentService(session_factory=default_session_factory)

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,95 +1,77 @@
-"""Tests for transcription.ui.jobs_page."""
+"""Tests for the jobs page route."""
 
+from pathlib import Path
 from uuid import uuid4
 
 import pytest
 
-from transcription.models import Document, Job, Transcript
+from transcription.models import JobStatus
-from transcription.ui.jobs_page import fetch_job_detail, fetch_jobs
 
 
 @pytest.mark.integration
-class TestJobsListBehavior:
+class TestPageRendering:
-    """Verify job list data and rendering helpers."""
+    """Verify jobs routes render correctly with real app wiring."""
 
-    def test_fetch_jobs_returns_job_view_rows(self, session, monkeypatch):
+    def test_jobs_page_renders_empty_state(self, app_client):
-        """fetch_jobs returns normalized JobView rows for UI consumption."""
+        """GET /ui/jobs renders the page and empty-state text when no jobs exist."""
-        document = Document(filename="letter.jpg", file_path="uploads/letter.jpg")
+        _, client = app_client
-        session.add(document)
+        response = client.get("/ui/jobs")
-        session.commit()
-        session.refresh(document)
 
-        job = Job(document_id=document.id)
+        assert response.status_code == 200
-        session.add(job)
+        assert "Transcription Jobs" in response.text
-        session.commit()
+        assert "No jobs yet." in response.text
 
-        class _SessionContext:
+    def test_jobs_page_lists_seeded_jobs(self, app_client, seed_job):
-            def __enter__(self):
+        """GET /ui/jobs lists seeded jobs from the in-memory database."""
-                return session
+        _, client = app_client
+        seed_job(filename="sample.pdf", status=JobStatus.TRANSCRIBED, transcript_text="done")
 
-            def __exit__(self, exc_type, exc, tb):
+        response = client.get("/ui/jobs")
-                return False
 
-        monkeypatch.setattr("transcription.ui.jobs_page.get_session", lambda: _SessionContext())
+        assert response.status_code == 200
+        assert "sample.pdf" in response.text
+        assert "transcribed" in response.text
 
-        rows = fetch_jobs()
+    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,
+        )
 
-        assert len(rows) == 1
+        response = client.get(f"/ui/jobs/{job_id}")
-        assert rows[0].id == job.id
-        assert rows[0].status == "queued"
 
+        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
 
-@pytest.mark.integration
+    def test_job_detail_page_rejects_invalid_id(self, app_client):
-class TestJobDetailBehavior:
+        """GET /ui/jobs/{job_id} shows validation feedback for malformed IDs."""
-    """Verify job detail retrieval behavior."""
+        _, client = app_client
+        response = client.get("/ui/jobs/not-a-uuid")
 
-    def test_fetch_job_detail_returns_related_records_when_present(self, session, monkeypatch):
+        assert response.status_code == 200
-        """fetch_job_detail returns job, document, and transcript when available."""
+        assert "Invalid job id" in response.text
-        document = Document(filename="typed.jpg", file_path="uploads/typed.jpg")
-        session.add(document)
-        session.commit()
-        session.refresh(document)
 
-        job = Job(document_id=document.id)
+    def test_job_detail_page_handles_missing_job(self, app_client):
-        session.add(job)
+        """GET /ui/jobs/{job_id} shows not-found state for unknown IDs."""
-        session.commit()
+        _, client = app_client
-        session.refresh(job)
+        missing_id = uuid4()
+        response = client.get(f"/ui/jobs/{missing_id}")
 
-        transcript = Transcript(job_id=job.id, text="Transcript text")
+        assert response.status_code == 200
-        session.add(transcript)
+        assert "Job not found" in response.text
-        session.commit()
-
-        class _SessionContext:
-            def __enter__(self):
-                return session
-
-            def __exit__(self, exc_type, exc, tb):
-                return False
-
-        monkeypatch.setattr("transcription.ui.jobs_page.get_session", lambda: _SessionContext())
-
-        fetched_job, fetched_document, fetched_transcript = fetch_job_detail(job.id)
-
-        assert fetched_job is not None
-        assert fetched_document is not None
-        assert fetched_transcript is not None
-        assert fetched_job.id == job.id
-        assert fetched_document.id == document.id
-        assert fetched_transcript.job_id == job.id
-
-    def test_fetch_job_detail_returns_nones_for_missing_job(self, session, monkeypatch):
-        """fetch_job_detail returns triple None when job does not exist."""
-        class _SessionContext:
-            def __enter__(self):
-                return session
-
-            def __exit__(self, exc_type, exc, tb):
-                return False
-
-        monkeypatch.setattr("transcription.ui.jobs_page.get_session", lambda: _SessionContext())
-
-        fetched_job, fetched_document, fetched_transcript = fetch_job_detail(uuid4())
-
-        assert fetched_job is None
-        assert fetched_document is None
-        assert fetched_transcript is None

tests/ui/test_pages_registration.py

@@ -1,26 +1,18 @@
 """Tests for UI page registration wiring."""
 
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
 import pytest
 
-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_adds_expected_routes(self):
+    def test_ui_mount_serves_registered_pages(self, app_client):
-        """register_pages wires upload and jobs routes into the app."""
+        """Mounted UI routes respond successfully when the full app is created."""
-        app = FastAPI()
+        _, client = app_client
-        register_pages(app)
-        app.add_api_route("/healthz", lambda: {"status": "ok"}, methods=["GET"])
 
-        client = TestClient(app)
+        upload_response = client.get("/ui/upload")
-        ui_response = client.get("/ui")
+        jobs_response = client.get("/ui/jobs")
-        health_response = client.get("/healthz")
 
-        assert ui_response.status_code == 200
+        assert upload_response.status_code == 200
-        assert health_response.status_code == 200
+        assert jobs_response.status_code == 200
-        assert health_response.json() == {"status": "ok"}

tests/ui/test_upload_page.py

@@ -1,53 +1,35 @@
-"""Tests for transcription.ui.upload_page."""
+"""Tests for upload and entry-point routes."""
-
-from pathlib import Path
-from uuid import uuid4
 
 import pytest
 
-from transcription.services.upload import UploadError, UploadJobResult
-from transcription.ui import upload_page
 
+@pytest.mark.integration
+class TestPageRendering:
+    """Verify upload-related routes return working pages."""
 
-@pytest.mark.unit
+    def test_root_redirects_to_ui(self, app_client):
-class TestUploadPageBehavior:
+        """GET / redirects to the UI mount point."""
-    """Verify upload page helper and submission behavior."""
+        _, client = app_client
+        response = client.get("/", follow_redirects=False)
 
-    def test_accepted_upload_types_contains_supported_extensions(self):
+        assert response.status_code == 307
-        """accepted_upload_types includes all MVP-supported upload extensions."""
+        assert response.headers["location"] == "/ui"
-        accepted = upload_page.accepted_upload_types()
-        assert ".jpg" in accepted
-        assert ".jpeg" in accepted
-        assert ".png" in accepted
-        assert ".tif" in accepted
-        assert ".tiff" in accepted
-        assert ".pdf" in accepted
 
-    def test_submit_upload_calls_upload_service(self, monkeypatch):
+    def test_ui_redirects_to_upload(self, app_client):
-        """submit_upload delegates file persistence and job creation to upload service."""
+        """GET /ui redirects to the upload page."""
-        expected = UploadJobResult(
+        _, client = app_client
-            document_id=uuid4(),
+        response = client.get("/ui", follow_redirects=False)
-            job_id=uuid4(),
-            stored_path=Path("uploads/mock.jpg"),
-            original_filename="mock.jpg",
-        )
 
-        def fake_create_upload_job(*, filename: str, file_bytes: bytes):
+        assert response.status_code == 307
-            assert filename == "mock.jpg"
+        assert response.headers["location"] == "/ui/upload"
-            assert file_bytes == b"bytes"
-            return expected
 
-        monkeypatch.setattr(upload_page, "create_upload_job", fake_create_upload_job)
+    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")
 
-        result = upload_page.submit_upload(filename="mock.jpg", file_bytes=b"bytes")
+        assert response.status_code == 200
-        assert result == expected
+        assert "Upload Document" in response.text
-
+        assert "Select document file" in response.text
-    def test_submit_upload_surfaces_upload_error(self, monkeypatch):
+        assert "Upload" in response.text
-        """submit_upload raises UploadError for invalid upload payloads."""
+        assert "Jobs" in response.text
-        def fake_create_upload_job(*, filename: str, file_bytes: bytes):
-            raise UploadError("invalid payload")
-
-        monkeypatch.setattr(upload_page, "create_upload_job", fake_create_upload_job)
-
-        with pytest.raises(UploadError):
-            upload_page.submit_upload(filename="bad.jpg", file_bytes=b"")