V6.1 UI refinements, add Maintenance jobs to Settings

docs/roadmap_plan.md

@@ -27,25 +27,24 @@ Detailed plan: [`v6_0_hosting_migration_plan.md`](v6_0_hosting_migration_plan.md
 
 ## V6.1 - Testing and Refinement
 
-[*Potential app refinements:*
+Objective: improve navigation and operational workflows after user feedback.
-* Create a **single “Maintenance Jobs” UI** backed by your existing async worker pattern, not direct shell execution from the page.
-  1. Add a `maintenance_run` table (`id`, `job_type`, `status`, `started_at`, `finished_at`, `triggered_by`, `summary`, `log_path`, `error_detail`).
-  2. In the UI, add two buttons: **Run Backup** and **Run Storage Reconciliation**; clicking creates a run row and enqueues worker execution.
-  3. Worker executes the existing scripts/commands, captures stdout/stderr to a timestamped log file, updates run status/summary.
-  4. UI shows a run history grid with live status, duration, summary, and “View Log”/“Download Log”.
-  5. Add optional schedule controls (daily/weekly) that create queued runs through the same path so manual and scheduled runs behave identically.
 
-* Consider a difference Sources for Document page
+### Scope
-  * Presenting the list of Document sources is not very useful.
+1. Make Document Detail the primary source-page workspace:
-  * Consider presenting a thumbnail gallery instead. I think this would make it easier to select the single source file the user is looking for.
+   - Use Source-style pan/zoom + previous/next page controls.
-  * This idea may be useful in other areas as well.
+   - Move editable revision controls into Document Detail.
-
+   - Move archival/system metadata to dedicated Document Info route.
-* People detail page -> LINKED DOCUMENTS:
+2. Simplify top navigation:
-  * How to handle long list of identically named documents (e.g., "Hig postcard to Zenna")? The table shown is not useful, but there isn't enough real estate in the column to add something like the document date, I don't think.
+   - Remove top-level Tags and Sources entries.
-
+   - Retire the Tags page and the global Source Asset Records entry flow.
-* Start reviewing what the UI looks like on a smart phone. How do those cards arrange themselves on a phone?
+3. Improve list/detail clarity:
-
+   - Add Document transcription status to Archival Documents list.
-]
+   - Add Document Date in People Detail -> Linked Documents table.
+4. Add worker-backed Settings maintenance runs:
+   - Add `maintenance_run` persistence (`id`, `job_type`, `status`, `started_at`, `finished_at`, `triggered_by`, `summary`, `log_path`, `error_detail`).
+   - Add Run Backup and Run Storage Reconciliation actions that enqueue runs and execute in the worker.
+   - Add run history with status, duration, summary, and log view/download.
+   - Defer daily/weekly scheduling controls to V6.2.
 
 ## V6.2 - Reporting Features
 

docs/schema.md

@@ -1,14 +1,15 @@
-# Data Model and Persistence Schema (Current Baseline: V5.1)
+# Data Model and Persistence Schema (Current Baseline: V6.1)
 
-This document is the field-accurate V5.1 schema contract aligned to `src/transcription/db/models.py`.
+This document is the field-accurate V6.1 schema contract aligned to `src/transcription/db/models.py`.
 
 ## Source of Truth Anchors
 
-- `src/transcription/db/models.py:60-78` (status and purpose enums)
+- `src/transcription/db/models.py` (status and purpose enums, including maintenance lifecycle enums)
 - `src/transcription/db/models.py:80-120` (`DocumentType`, `PersonRole`)
 - `src/transcription/db/models.py:122-172` (`Tag`, `Document`)
 - `src/transcription/db/models.py:175-281` (`Person`, `Photo`, `DocumentPerson`, `DocumentTag`)
 - `src/transcription/db/models.py:285-347` (`Job`)
+- `src/transcription/db/models.py` (`MaintenanceRun`)
 - `src/transcription/db/models.py:350-462` (`Source`, `JobSource`)
 - `src/transcription/db/models.py:465-522` (`ExecutionAttempt`)
 
@@ -30,6 +31,9 @@ erDiagram
     Job ||--o{ JobSource : includes
     Source ||--o{ JobSource : participates
     JobSource ||--o{ ExecutionAttempt : attempts
+    MaintenanceRun {
+        uuid id PK
+    }
 ```
 
 ## Authoritative Enumerations
@@ -54,6 +58,18 @@ erDiagram
 - `transcription`
 - `retranscription`
 
+### MaintenanceJobType
+
+- `backup`
+- `storage_reconciliation`
+
+### MaintenanceRunStatus
+
+- `queued`
+- `processing`
+- `succeeded`
+- `failed`
+
 ## Field-Accurate Table Contracts
 
 ### `DocumentType`
@@ -201,6 +217,22 @@ Constraint:
 Index:
 - `Index("ix_job_status_date_created", "status", "date_created")`
 
+### `MaintenanceRun`
+
+| Field | Type | Notes |
+| :--- | :--- | :--- |
+| `id` | `UUID` | PK |
+| `job_type` | `MaintenanceJobType` | non-null enum |
+| `status` | `MaintenanceRunStatus` | non-null enum, default `queued` |
+| `started_at` | `datetime \| None` | optional |
+| `finished_at` | `datetime \| None` | optional |
+| `triggered_by` | `str \| None` | optional |
+| `summary` | `str \| None` | optional |
+| `log_path` | `str \| None` | optional, log-root-relative POSIX path |
+| `error_detail` | `str \| None` | optional internal failure detail |
+| `created_at` | `datetime` | default now |
+| `updated_at` | `datetime` | default now, onupdate |
+
 ### `Source`
 
 | Field | Type | Notes |

docs/ui/pages/documents.md

@@ -11,10 +11,11 @@ Documents manages the archival record for each historical artifact independently
 | `/documents` | Searchable archival Document list. |
 | `/documents/new` | Create a Document. |
 | `/documents/{document_id}` | View one Document and its related records. |
+| `/documents/{document_id}/info` | View archival metadata and system logistics for one Document. |
 | `/documents/{document_id}/edit` | Edit metadata and the complete Linked People set. |
 | `/documents/{document_id}/delete` | Confirm or block deletion. |
 | `/documents/{document_id}/jobs` | Show Jobs belonging to the Document. |
-| `/documents/{document_id}/sources` | Redirect to the Document-filtered Sources list. |
+| `/documents/{document_id}/sources` | Redirect back to Document Detail. |
 | `/documents/{document_id}/print` | Preview and browser-print the persisted Document. |
 
 ## List Behavior
@@ -22,10 +23,11 @@ Documents manages the archival record for each historical artifact independently
 - The title is **Archival Documents**.
 - **Create new document** opens the create route.
 - The table defaults to Document Title order and supports search and column sorting.
-- Columns are Document Title, Author, Tags, Document Date, Type, and # Sources.
+- Columns are Document Title, Author, Tags, Document Date, Type, # Sources, and Transcription Status.
 - Document Title is left-aligned; the remaining columns are centered.
 - Author lists all linked people in the `author` role.
 - # Sources reflects the count of linked Source rows for each Document.
+- Transcription Status reflects the most recent Job status for that Document; documents with no Jobs show a blank marker.
 - Date display prefers exact date, then approximate date, then `Unknown`.
 - Selecting a row opens Document Detail.
 - Row navigation includes list context so Document Detail provides **Back to Documents**.
@@ -70,14 +72,19 @@ Rules:
 
 - The heading shows name, type, and internal ID.
 - The header includes a contextual back action: **Back to Documents** by default, **Back to Person** when opened from Person Detail, and **Back to Job** when opened from Job Detail.
-- The first Source, when present, appears in the dark-room viewer.
+- The detail workspace shows a Source-style pan/zoom media viewer with **Previous Page** / **Next Page** navigation for document source pages.
-- Archival Metadata shows authors, Document Type, tags, Document date (`MM-DD-YYYY` for exact dates), location (linked to Google Maps when present), and archive identifier. Notes appear in a separate archival-notes block within the same card.
+- The center column is **Editable Revision** for the active source page.
-- System Logistics shows created and updated timestamps.
 - Related People are grouped by role and link to Person Detail.
-- **Sources & Pipeline Jobs** shows counts and actions for filtered Sources, Document Jobs, and adding a Job.
+- **Source Pages & Transcriptions** shows source/job counts and actions for source detail, document jobs, and adding a Job.
-- **Edit Document**, **Print**, and **Delete** are available from the header.
+- **Edit Document**, **Print**, **Document Details**, and **Delete** are available from the header.
 - Invalid IDs and missing Documents produce explicit states without rendering a partial page.
 
+## Document Info Behavior
+
+- `/documents/{document_id}/info` contains **Archival Metadata** and **System Logistics**.
+- It includes a **Back to Document** action.
+- Archival metadata includes authors, document type, tags, document date, location (linked when present), archive identifier, and notes.
+
 ## Print Behavior
 
 - Print opens a dedicated preview for persisted Document data.

docs/ui/pages/people.md

@@ -64,7 +64,7 @@ Rules:
 - Birth and death place values are clickable links to Google Maps when present.
 - FamilySearch ID is shown as a metadata value and is clickable to the FamilySearch person details route when present.
 - Biography has an explicit empty value.
-- Linked Documents render as a table with **Document Name**, **Role**, and **Number of Pages**; selecting a row opens Document Detail.
+- Linked Documents render as a table with **Document Name**, **Document Date**, **Role**, and **Number of Pages**; selecting a row opens Document Detail.
 - No links shows both an empty state and guidance to link from a Document workflow.
 - System Logistics shows created and updated timestamps.
 

docs/ui/pages/settings.md

@@ -8,7 +8,7 @@ Settings manages installation-local registries, safe runtime .env settings, and
 
 | Route | Purpose |
 | --- | --- |
-| `/settings` | Manage Runtime Settings, Document Types, Person Roles, Tags, Prompts, and Home Page Text. |
+| `/settings` | Manage Runtime Settings, Document Types, Person Roles, Tags, Prompts, Home Page Text, and Maintenance runs. |
 
 ## Behavior
 
@@ -19,6 +19,7 @@ Settings manages installation-local registries, safe runtime .env settings, and
   - **Tags**
   - **Prompts**
   - **Home Page Text**
+  - **Maintenance**
   - **Runtime Settings**
 - Runtime Settings exposes an allowlisted set of non-secret fields synchronized with `Settings` model fields except excluded secret/unsafe fields.
 - Runtime Settings is rendered as a compact two-column editor (**Setting**, **Value**) in a centered, narrower responsive container.
@@ -31,10 +32,13 @@ Settings manages installation-local registries, safe runtime .env settings, and
 - Document Types, Person Roles, and Tags support Add/Edit/Delete with existing guardrails.
 - Prompts exposes only `transcribe_document.md` for editing and restore-from-backup.
 - Home Page Text edits the same Markdown content rendered on `/homepage`.
+- Maintenance provides queue-backed **Run Backup** and **Run Storage Reconciliation** actions.
+- Maintenance run history shows job type, status, started/finished timestamps, duration, summary, and log view/download actions.
+- Maintenance actions enqueue work and signal the worker; the page itself does not execute shell commands directly.
 
 ## Acceptance Checklist
 
-- `/ui/settings` renders all six tabs.
+- `/ui/settings` renders all seven tabs.
 - Registry and prompt workflows keep existing validation and error handling.
 - Runtime Settings excludes secret fields and rejects invalid values.
 - Saving Home Page Text persists content for the homepage view.

docs/ui/pages/sources.md

@@ -8,7 +8,7 @@ Sources manages individual archived page/file records. It provides source-media
 
 | Route | Purpose |
 | --- | --- |
-| `/sources` | Global or filtered Source list. |
+| `/sources` | Document-filtered or Job-filtered Source list; global route redirects to Documents. |
 | `/sources/{source_id}` | View media, transcription, revision, metadata, and evidence. |
 | `/sources/{source_id}/delete` | Confirm or block deletion. |
 
@@ -16,8 +16,8 @@ The list accepts optional `document_id` and `job_id` query parameters. Document
 
 ## List Behavior
 
-- The title is **Source Asset Records**, **Sources for Document**, or **Sources for Job** according to context.
+- The global `/sources` route redirects to `/documents`.
-- Global context provides **Create Job**.
+- Filtered list titles are **Sources for Document** and **Sources for Job**.
 - Filtered context provides **Back to Document** or **Back to Job**.
 - Rows are ordered by page number and then upload name.
 - Columns are Upload Title, Page Number, Document Name, Status, and Error Detail.
@@ -30,7 +30,7 @@ The list accepts optional `document_id` and `job_id` query parameters. Document
 ## Detail Behavior
 
 - The heading shows page number, upload name, and Source ID.
-- **Back to Sources** returns to the global list.
+- **Back to Document** returns to Document Detail for the active source page.
 - **Retranscribe Source** opens Create Processing Job with this Source and its Document locked.
 - **Delete Source** opens the guarded delete route.
 - Previous and Next navigate only among Sources belonging to the same Document in page order; unavailable boundary actions are disabled.

docs/ui/pages/tags.md

@@ -1,31 +0,0 @@
-# Tags Page Contract
-
-## Purpose
-
-Tags provides a dedicated browse/filter entry point for document tagging workflows.
-
-## Route
-
-| Route | Purpose |
-| --- | --- |
-| `/tags` | Browse Documents grouped by Tag and filter to one Tag. |
-
-## Behavior
-
-- The page title is **Tags**.
-- When no tags exist, the page shows `No tags are configured yet.`
-- A Tag filter select allows narrowing to one tag.
-- Each rendered group header includes the tag label and document count.
-- Document names are clickable and open Document Detail.
-
-## Acceptance Checklist
-
-- `/ui/tags` renders successfully from the main navigation.
-- Group counts match the number of linked Documents per Tag.
-- Filtering hides non-matching tag groups.
-
-## Implementation Anchors
-
-- `src/transcription/ui/pages/tags_page.py`
-- `src/transcription/services/documents.py`
-- `tests/ui/test_tags_page.py`

src/transcription/db/models.py

@@ -82,6 +82,18 @@ class JobPurpose(StrEnum):
     RETRANSCRIPTION = "retranscription"
 
 
+class MaintenanceJobType(StrEnum):
+    BACKUP = "backup"
+    STORAGE_RECONCILIATION = "storage_reconciliation"
+
+
+class MaintenanceRunStatus(StrEnum):
+    QUEUED = "queued"
+    PROCESSING = "processing"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+
+
 class DocumentType(SQLModel, table=True):
     """Registry of allowed document types."""
 
@@ -390,6 +402,47 @@ class Job(SQLModel, table=True):
         return "unknown"
 
 
+class MaintenanceRun(SQLModel, table=True):
+    """A queued/processed maintenance task execution record."""
+
+    __tablename__ = "maintenance_run"
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True)
+    job_type: MaintenanceJobType = Field(
+        sa_column=Column(
+            SAEnum(
+                MaintenanceJobType,
+                values_callable=lambda enum_cls: [item.value for item in enum_cls],
+                native_enum=False,
+            ),
+            nullable=False,
+        )
+    )
+    status: MaintenanceRunStatus = Field(
+        default=MaintenanceRunStatus.QUEUED,
+        sa_column=Column(
+            SAEnum(
+                MaintenanceRunStatus,
+                values_callable=lambda enum_cls: [item.value for item in enum_cls],
+                native_enum=False,
+            ),
+            nullable=False,
+            default=MaintenanceRunStatus.QUEUED.value,
+        ),
+    )
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    triggered_by: str | None = None
+    summary: str | None = None
+    log_path: str | None = None
+    error_detail: str | None = None
+    created_at: datetime = Field(default_factory=_utc_now_naive)
+    updated_at: datetime = Field(
+        default_factory=_utc_now_naive,
+        sa_column_kwargs={"onupdate": _utc_now_naive},
+    )
+
+
 class Source(SQLModel, table=True):
     """A document source image or PDF page."""
 

src/transcription/services/__init__.py

@@ -11,6 +11,7 @@ from ..config import Settings
 from .documents import DocumentService
 from .evidence import EvidenceService
 from .jobs import JobService
+from .maintenance import MaintenanceService
 from .people import PeopleService
 from .photos import PhotosService
 from .prompts import PromptStore
@@ -20,6 +21,7 @@ __all__ = [
     "DocumentService",
     "EvidenceService",
     "JobService",
+    "MaintenanceService",
     "PeopleService",
     "PhotosService",
     "PromptStore",
@@ -35,6 +37,7 @@ class ServiceBundle:
     documents: DocumentService = field(default_factory=DocumentService)
     sources: SourceService = field(default_factory=SourceService)
     jobs: JobService = field(default_factory=JobService)
+    maintenance: MaintenanceService = field(default_factory=MaintenanceService)
     people: PeopleService = field(default_factory=PeopleService)
     photos: PhotosService = field(default_factory=PhotosService)
     evidence: EvidenceService = field(default_factory=EvidenceService)
@@ -53,6 +56,7 @@ class ServiceBundle:
             documents=DocumentService(session_factory=session_factory, settings=settings),
             sources=SourceService(session_factory=session_factory, settings=settings),
             jobs=JobService(session_factory=session_factory, settings=settings),
+            maintenance=MaintenanceService(session_factory=session_factory, settings=settings),
             people=PeopleService(session_factory=session_factory, settings=settings),
             photos=PhotosService(session_factory=session_factory, settings=settings),
             evidence=EvidenceService(session_factory=session_factory, settings=settings),

src/transcription/services/documents.py

@@ -303,6 +303,7 @@ class DocumentService(ServiceBase):
                 selectinload(Document.document_type_ref),
                 selectinload(Document.document_tags).selectinload(orm_attribute(DocumentTag.tag_ref)),
                 selectinload(Document.sources),
+                selectinload(Document.jobs),
             )
             result = await _session.exec(query)
             return result.all()

src/transcription/services/maintenance.py

@@ -0,0 +1,323 @@
+"""Queue-backed maintenance operations executed by the worker loop."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
+from pathlib import Path
+from uuid import UUID
+
+from sqlalchemy import update
+from sqlmodel import col
+from sqlmodel import func
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from transcription.db.models import Document
+from transcription.db.models import MaintenanceJobType
+from transcription.db.models import MaintenanceRun
+from transcription.db.models import MaintenanceRunStatus
+from transcription.db.models import Source
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.errors import classify_unexpected_error
+
+from .base import ServiceBase
+
+
+def _utc_now_naive() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+@dataclass(frozen=True, slots=True)
+class MaintenanceExecution:
+    """In-memory result for one executed maintenance run."""
+
+    status: MaintenanceRunStatus
+    summary: str
+    output: str
+    error_detail: str | None = None
+
+
+class MaintenanceError(AppError):
+    """Raised when maintenance operations cannot be enqueued or executed."""
+
+
+class MaintenanceService(ServiceBase):
+    """Persist and execute background maintenance runs."""
+
+    async def list_runs(
+        self,
+        *,
+        limit: int = 100,
+        session: AsyncSession | None = None,
+    ) -> list[MaintenanceRun]:
+        async with self._session_scope(session) as _session:
+            query = (
+                select(MaintenanceRun)
+                .order_by(col(MaintenanceRun.created_at).desc(), col(MaintenanceRun.id).desc())
+                .limit(limit)
+            )
+            return list((await _session.exec(query)).all())
+
+    async def enqueue_run(
+        self,
+        *,
+        job_type: MaintenanceJobType,
+        triggered_by: str = "ui.settings",
+        session: AsyncSession | None = None,
+    ) -> MaintenanceRun:
+        run = MaintenanceRun(
+            job_type=job_type,
+            status=MaintenanceRunStatus.QUEUED,
+            triggered_by=triggered_by,
+        )
+        async with self._session_scope(session) as _session:
+            _session.add(run)
+            await self._finalize(session=_session, caller_session=session, refresh=(run,))
+        return run
+
+    async def claim_next_queued_run(self, *, session: AsyncSession | None = None) -> MaintenanceRun | None:
+        async with self._session_scope(session) as _session:
+            now = _utc_now_naive()
+            queued_run_id = (
+                select(col(MaintenanceRun.id))
+                .where(col(MaintenanceRun.status) == MaintenanceRunStatus.QUEUED)
+                .order_by(col(MaintenanceRun.created_at), col(MaintenanceRun.id))
+                .limit(1)
+                .scalar_subquery()
+            )
+            claim_statement = (
+                update(MaintenanceRun)
+                .where(col(MaintenanceRun.id) == queued_run_id)
+                .where(col(MaintenanceRun.status) == MaintenanceRunStatus.QUEUED)
+                .values(
+                    status=MaintenanceRunStatus.PROCESSING,
+                    started_at=now,
+                    updated_at=now,
+                )
+                .returning(col(MaintenanceRun.id))
+            )
+            claimed_row = (await _session.exec(claim_statement)).first()
+            if claimed_row is None:
+                return None
+            claimed_run_id = claimed_row if isinstance(claimed_row, UUID) else claimed_row[0]
+            run = await _session.get(MaintenanceRun, claimed_run_id)
+            if run is None:
+                return None
+            await self._finalize(session=_session, caller_session=session, refresh=(run,))
+            return run
+
+    async def process_next_queued_run(self, *, session: AsyncSession | None = None) -> bool:
+        run = await self.claim_next_queued_run(session=session)
+        if run is None:
+            return False
+
+        execution = await self._execute_run(run)
+        await self._finalize_run(run_id=run.id, execution=execution, session=session)
+        return True
+
+    async def _finalize_run(
+        self,
+        *,
+        run_id: UUID,
+        execution: MaintenanceExecution,
+        session: AsyncSession | None = None,
+    ) -> None:
+        now = _utc_now_naive()
+        log_path = self._write_log(run_id=run_id, output=execution.output)
+        async with self._session_scope(session) as _session:
+            run = await _session.get(MaintenanceRun, run_id)
+            if run is None:
+                raise MaintenanceError(
+                    "Maintenance run not found while finalizing.",
+                    category=ErrorCategory.NOT_FOUND,
+                    suggestion="Refresh the page and retry.",
+                )
+            run.status = execution.status
+            run.summary = execution.summary
+            run.error_detail = execution.error_detail
+            run.log_path = log_path
+            run.finished_at = now
+            run.updated_at = now
+            await self._finalize(session=_session, caller_session=session, refresh=(run,))
+
+    async def _execute_run(self, run: MaintenanceRun) -> MaintenanceExecution:
+        if run.job_type == MaintenanceJobType.BACKUP:
+            return await self._execute_backup()
+        if run.job_type == MaintenanceJobType.STORAGE_RECONCILIATION:
+            return await self._execute_storage_reconciliation()
+        raise MaintenanceError(
+            "Unsupported maintenance job type.",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Choose a supported maintenance action and retry.",
+        )
+
+    async def _execute_backup(self) -> MaintenanceExecution:
+        script_path = Path("deploy") / "backup" / "create_postgres_backup.sh"
+        if not script_path.is_file():
+            return MaintenanceExecution(
+                status=MaintenanceRunStatus.FAILED,
+                summary="Backup script is unavailable in this environment.",
+                output="Backup script not found.",
+                error_detail=f"Missing script: {script_path}",
+            )
+
+        command = ["sh", str(script_path)]
+        try:
+            process = await asyncio.create_subprocess_exec(
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.STDOUT,
+            )
+            stdout, _ = await process.communicate()
+        except OSError as exc:
+            return MaintenanceExecution(
+                status=MaintenanceRunStatus.FAILED,
+                summary="Backup command failed to start.",
+                output=f"Failed to execute {' '.join(command)}",
+                error_detail=f"{type(exc).__name__}: {exc}",
+            )
+
+        output = stdout.decode("utf-8", errors="replace")
+        if process.returncode == 0:
+            return MaintenanceExecution(
+                status=MaintenanceRunStatus.SUCCEEDED,
+                summary="Backup completed successfully.",
+                output=output,
+            )
+        return MaintenanceExecution(
+            status=MaintenanceRunStatus.FAILED,
+            summary="Backup command failed.",
+            output=output,
+            error_detail=f"Exit code: {process.returncode}",
+        )
+
+    async def _execute_storage_reconciliation(self) -> MaintenanceExecution:
+        try:
+            mismatches = await self._collect_storage_mismatches()
+        except Exception as exc:  # noqa: BLE001
+            error = classify_unexpected_error(exc, operation="maintenance.storage_reconciliation")
+            return MaintenanceExecution(
+                status=MaintenanceRunStatus.FAILED,
+                summary="Storage reconciliation failed.",
+                output="Storage reconciliation failed before completion.",
+                error_detail=error.detail,
+            )
+
+        if mismatches:
+            report = "\n".join(f"- {item}" for item in mismatches)
+            return MaintenanceExecution(
+                status=MaintenanceRunStatus.FAILED,
+                summary=f"Storage reconciliation found {len(mismatches)} issue(s).",
+                output=report,
+                error_detail="Reconciliation mismatches were detected.",
+            )
+
+        return MaintenanceExecution(
+            status=MaintenanceRunStatus.SUCCEEDED,
+            summary="Storage reconciliation found no mismatches.",
+            output="No storage reconciliation mismatches detected.",
+        )
+
+    async def _collect_storage_mismatches(self) -> list[str]:
+        upload_root = self.settings.upload_dir
+        folder_ids = _document_folder_ids(upload_root)
+        doc_ids = await self._document_ids()
+        source_counts = await self._source_counts_by_document()
+        folder_by_normalized = {_normalize_identifier(folder_id): folder_id for folder_id in folder_ids}
+        doc_by_normalized = {_normalize_identifier(doc_id): doc_id for doc_id in doc_ids}
+        source_counts_by_normalized = {
+            _normalize_identifier(document_id): count for document_id, count in source_counts.items()
+        }
+        mismatches: list[str] = []
+
+        missing_in_table = sorted(set(folder_by_normalized) - set(doc_by_normalized))
+        for folder_key in missing_in_table:
+            folder_name = folder_by_normalized[folder_key]
+            mismatches.append(f"document-folder-without-row: documents/{folder_name}")
+
+        missing_in_folders = sorted(set(doc_by_normalized) - set(folder_by_normalized))
+        for doc_key in missing_in_folders:
+            doc_id = doc_by_normalized[doc_key]
+            source_count = source_counts_by_normalized.get(doc_key, 0)
+            mismatches.append(f"document-row-without-folder: {doc_id} (source rows: {source_count})")
+
+        for doc_key in sorted(doc_by_normalized):
+            doc_id = doc_by_normalized[doc_key]
+            folder_name = folder_by_normalized.get(doc_key)
+            db_count = source_counts_by_normalized.get(doc_key, 0)
+            file_count = _source_file_count_for_document(upload_root, folder_name) if folder_name is not None else 0
+            if db_count != file_count:
+                path_label = f"documents/{folder_name}" if folder_name is not None else "documents/<missing-folder>"
+                mismatches.append(
+                    f"source-count-mismatch: {doc_id} -> source rows: {db_count}, files in {path_label}: {file_count}"
+                )
+        return mismatches
+
+    async def _document_ids(self) -> set[str]:
+        async with self._session_scope() as session:
+            rows = await session.exec(select(Document.id))
+            return {str(item) for item in rows.all()}
+
+    async def _source_counts_by_document(self) -> dict[str, int]:
+        async with self._session_scope() as session:
+            rows = await session.exec(
+                select(
+                    Source.document_id,
+                    func.count(Source.id),  # ty: ignore[invalid-argument-type] - SQLAlchemy descriptor false positive.
+                ).group_by(
+                    Source.document_id  # ty: ignore[invalid-argument-type] - SQLAlchemy descriptor false positive.
+                )
+            )
+            return {str(document_id): int(count) for document_id, count in rows}
+
+    def read_log_bytes(self, *, log_path: str) -> bytes:
+        candidate = (self.settings.log_dir / Path(log_path)).resolve()
+        base = self.settings.log_dir.resolve()
+        try:
+            candidate.relative_to(base)
+        except ValueError as exc:
+            raise MaintenanceError(
+                "Maintenance log path is invalid.",
+                category=ErrorCategory.VALIDATION,
+                suggestion="Refresh and retry.",
+                detail=f"Requested path outside log root: {candidate}",
+            ) from exc
+        if not candidate.is_file():
+            raise MaintenanceError(
+                "Maintenance log file is unavailable.",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Refresh and retry.",
+            )
+        return candidate.read_bytes()
+
+    def _write_log(self, *, run_id: UUID, output: str) -> str:
+        timestamp = _utc_now_naive().strftime("%Y%m%d-%H%M%S")
+        logs_dir = self.settings.log_dir / "maintenance"
+        logs_dir.mkdir(parents=True, exist_ok=True)
+        file_path = logs_dir / f"maintenance-{run_id}-{timestamp}.log"
+        file_path.write_text(output, encoding="utf-8")
+        return str(file_path.relative_to(self.settings.log_dir).as_posix())
+
+
+def _normalize_identifier(value: str) -> str:
+    return value.replace("-", "").strip().lower()
+
+
+def _document_folder_ids(root: Path) -> set[str]:
+    documents_root = root / "documents"
+    if not documents_root.exists():
+        return set()
+    return {entry.name for entry in documents_root.iterdir() if entry.is_dir()}
+
+
+def _source_file_count_for_document(root: Path, document_id: str | None) -> int:
+    if document_id is None:
+        return 0
+    directory = root / "documents" / document_id
+    if not directory.exists():
+        return 0
+    return sum(1 for entry in directory.iterdir() if entry.is_file())

src/transcription/ui/__init__.py

@@ -13,7 +13,6 @@ from transcription.ui.pages.people_page import register_page as register_people_
 from transcription.ui.pages.print_preview_page import register_page as register_print_preview_page
 from transcription.ui.pages.settings_page import register_page as register_settings_page
 from transcription.ui.pages.sources_page import register_page as register_sources_page
-from transcription.ui.pages.tags_page import register_page as register_tags_page
 from transcription.ui.resources import read_css
 from transcription.ui.theme import apply_archival_theme
 
@@ -37,7 +36,6 @@ def register_pages(app: FastAPI) -> None:
     """Register all NiceGUI pages and mount them onto the FastAPI app."""
     register_home_page()
     register_documents_page()
-    register_tags_page()
     register_people_page()
     register_print_preview_page()
     register_sources_page()

src/transcription/ui/components/app_shell.py

@@ -8,9 +8,7 @@ from transcription.ui.resources import read_svg
 
 NAV_ITEMS: tuple[tuple[str, str, str], ...] = (
     ("Documents", "/documents", "description"),
-    ("Tags", "/tags", "sell"),
     ("People", "/people", "group"),
-    ("Sources", "/sources", "folder"),
     ("Jobs", "/jobs", "work_history"),
     ("Settings", "/settings", "settings"),
 )
@@ -23,10 +21,6 @@ def _is_active_path(*, current_path: str, item_path: str) -> bool:
         return current_path == "/documents" or current_path.startswith("/documents/")
     if item_path == "/people":
         return current_path == "/people" or current_path.startswith("/people/")
-    if item_path == "/tags":
-        return current_path == "/tags" or current_path.startswith("/tags/")
-    if item_path == "/sources":
-        return current_path == "/sources" or current_path.startswith("/sources/")
     if item_path == "/settings":
         return current_path == "/settings" or current_path.startswith("/settings/")
     return current_path == item_path

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

@@ -25,6 +25,7 @@ class DocumentTableRow:
     tags: str
     document_date: str
     source_count: int
+    transcription_status: str | None = None
 
 
 def _serialize_rows(rows: Sequence[DocumentTableRow]) -> list[dict[str, Any]]:
@@ -37,6 +38,7 @@ def _serialize_rows(rows: Sequence[DocumentTableRow]) -> list[dict[str, Any]]:
             "tags": row.tags or "Not tagged",
             "document_date": row.document_date,
             "source_count": row.source_count,
+            "transcription_status": (row.transcription_status or "").lower(),
         }
         for row in rows
     ]
@@ -106,6 +108,15 @@ def render_documents_table(rows: Sequence[DocumentTableRow]) -> None:
                 "align": "center",
                 "style": "width: 10%;",
             },
+            {
+                "name": "transcription_status",
+                "label": "Transcription Status",
+                "field": "transcription_status",
+                "sortable": True,
+                "classes": "font-mono",
+                "align": "center",
+                "style": "width: 15%;",
+            },
         ],
         default_sort_by="name",
         search_placeholder="Search documents by title, type, or author...",
@@ -128,3 +139,20 @@ def render_documents_table(rows: Sequence[DocumentTableRow]) -> None:
         </q-td>
         """,
     )
+    table.add_slot(
+        "body-cell-transcription_status",
+        r"""
+        <q-td :props="props">
+            <span v-if="!props.value">-</span>
+            <q-chip
+                v-else
+                dense
+                square
+                size="sm"
+                :class="`ui-status ui-status--${props.value}`"
+            >
+                {{ props.value.toUpperCase() }}
+            </q-chip>
+        </q-td>
+        """,
+    )

src/transcription/ui/pages/documents_page.py

@@ -13,12 +13,14 @@ from nicegui import ui
 
 from transcription.config import Settings
 from transcription.db.models import Document
+from transcription.db.models import Source
 from transcription.db.registries import AUTHOR_ROLE_SEMANTIC_KEY
 from transcription.errors import ErrorCategory
 from transcription.services.documents import DocumentDeleteBlockedError
 from transcription.services.documents import DocumentError
 from transcription.services.documents import DocumentService
 from transcription.services.people import PeopleService
+from transcription.services.sources import SourceService
 from transcription.services.workflows import create_document_with_people
 from transcription.services.workflows import update_document_with_people
 from transcription.ui.components.app_shell import render_navigation_header
@@ -29,6 +31,7 @@ from transcription.ui.components.confirm_delete import render_delete_blocked_not
 from transcription.ui.components.data_display import archival_badge
 from transcription.ui.components.data_display import metadata_link_row
 from transcription.ui.components.data_display import metadata_row
+from transcription.ui.components.document_panzoom import render_document_panzoom
 from transcription.ui.components.error_presenter import run_ui_action
 from transcription.ui.components.error_presenter import show_error
 from transcription.ui.components.formatters import compact_date
@@ -45,7 +48,6 @@ from transcription.ui.components.primitives import render_empty_state
 from transcription.ui.components.primitives import section_header_row
 from transcription.ui.components.table.documents import DocumentTableRow
 from transcription.ui.components.table.documents import render_documents_table
-from transcription.ui.components.viewers import dark_room_viewer
 from transcription.ui.runtime import resolve_runtime_settings
 from transcription.ui.theme import page_header
 
@@ -201,6 +203,7 @@ def register_page() -> None:  # noqa: PLR0915
                     document_date=compact_date(doc.document_date, doc.document_date_raw),
                     document_type=(doc.document_type_ref.label if doc.document_type_ref is not None else ""),
                     source_count=len(doc.sources),
+                    transcription_status=_latest_job_status(doc),
                 )
                 for doc in documents
             ]
@@ -209,6 +212,7 @@ def register_page() -> None:  # noqa: PLR0915
     @ui.page("/documents/{document_id}")
     async def document_detail_page(request: Request, document_id: str, session_factory: SessionFactoryDep) -> None:
         document_service = DocumentService(session_factory=session_factory)
+        sources_service = SourceService(session_factory=session_factory)
         render_navigation_header(current_path="/documents")
         settings = resolve_runtime_settings(request)
         back_label = "Back to Documents"
@@ -238,6 +242,8 @@ def register_page() -> None:  # noqa: PLR0915
             show_error(exc, title="Load failed", operation="documents.read")
             return
 
+        active_source = _resolve_active_source(document, parse_uuid(request.query_params.get("source_id")))
+
         with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
             type_display = document.document_type_ref.label if document.document_type_ref is not None else "Unspecified"
             with section_header_row():
@@ -255,6 +261,11 @@ def register_page() -> None:  # noqa: PLR0915
                         on_click=lambda: ui.navigate.to(f"/documents/{document.id}/edit"),
                         icon="edit",
                     ).classes("ui-btn-primary text-xs")
+                    ui.button(
+                        "Document Details",
+                        on_click=lambda: ui.navigate.to(f"/documents/{document.id}/info"),
+                        icon="info",
+                    ).props("flat").classes("text-xs")
                     destructive_button(
                         "Delete",
                         on_click=lambda: ui.navigate.to(f"/documents/{document.id}/delete"),
@@ -263,10 +274,46 @@ def register_page() -> None:  # noqa: PLR0915
                     )
 
             with ui.grid().classes("w-full grid-cols-12 gap-4"):
-                _render_bento_viewer_zone(document, base_url=str(request.base_url), settings=settings)
+                _render_document_detail_viewer_zone(
-                _render_bento_metadata_zone(document)
+                    document=document,
+                    active_source=active_source,
+                    base_url=str(request.base_url),
+                    settings=settings,
+                )
+                _render_document_detail_revision_zone(
+                    source=active_source,
+                    sources_service=sources_service,
+                )
                 _render_bento_relations_zone(document)
 
+    @ui.page("/documents/{document_id}/info")
+    async def document_info_page(document_id: str, session_factory: SessionFactoryDep) -> None:
+        document_service = DocumentService(session_factory=session_factory)
+        render_navigation_header(current_path="/documents")
+
+        parsed_doc_id = parsed_record_id(document_id, noun="Document")
+        if parsed_doc_id is None:
+            return
+
+        try:
+            document = await document_service.read_document_detail(document_id=parsed_doc_id)
+        except DocumentError:
+            render_record_not_found("Document")
+            return
+        except Exception as exc:  # noqa: BLE001
+            show_error(exc, title="Load failed", operation="documents.info.read")
+            return
+
+        with ui.column().classes("w-full max-w-6xl mx-auto p-4 gap-4"):
+            with section_header_row():
+                page_header("Document Info", subtitle=f"{document.name} ({document.id})")
+                ui.button(
+                    "Back to Document",
+                    on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
+                    icon="arrow_back",
+                ).props("flat")
+            _render_bento_metadata_zone(document)
+
     @ui.page("/documents/{document_id}/jobs")
     async def document_jobs_page(document_id: str, session_factory: SessionFactoryDep) -> RedirectResponse:
         _ = session_factory
@@ -275,7 +322,7 @@ def register_page() -> None:  # noqa: PLR0915
     @ui.page("/documents/{document_id}/sources")
     async def document_sources_page(document_id: str, session_factory: SessionFactoryDep) -> RedirectResponse:
         _ = session_factory
-        return RedirectResponse(url=f"/ui/sources?document_id={document_id}")
+        return RedirectResponse(url=f"/ui/documents/{document_id}")
 
     @ui.page("/documents/{document_id}/edit")
     async def document_edit_page(document_id: str, session_factory: SessionFactoryDep) -> None:
@@ -539,27 +586,130 @@ def _render_document_form_fields(
     )
 
 
-def _render_bento_viewer_zone(document: Document, *, base_url: str, settings: Settings) -> None:
+def _resolve_active_source(document: Document, requested_source_id: UUID | None) -> Source | None:
-    with ui.column().classes("col-span-12 lg:col-span-4"):
+    ordered = sorted(
-        source_path = _first_source_path(document)
-        source_url = resolve_media_url(source_path, upload_dir=settings.upload_dir, base_url=base_url)
-        dark_room_viewer(source_url, count_label=f"{len(document.sources)} Source(s) Linked")
-
-
-def _first_source_path(document: Document) -> str | None:
-    if not document.sources:
-        return None
-    first_source = min(
         document.sources,
         key=lambda source: (source.page_number, source.upload_name.casefold(), source.filename.casefold()),
     )
-    return first_source.file_path
+    if not ordered:
+        return None
+    if requested_source_id is None:
+        return ordered[0]
+    for source in ordered:
+        if source.id == requested_source_id:
+            return source
+    return ordered[0]
+
+
+def _render_document_detail_viewer_zone(
+    *,
+    document: Document,
+    active_source: Source | None,
+    base_url: str,
+    settings: Settings,
+) -> None:
+    with ui.column().classes("col-span-12 lg:col-span-4 gap-2"):
+        _render_document_source_navigation(document=document, active_source=active_source)
+        if active_source is None:
+            render_document_panzoom(media_url=None, filename="No source pages", count_label="0 Source Pages")
+            return
+        source_url = resolve_media_url(active_source.file_path, upload_dir=settings.upload_dir, base_url=base_url)
+        render_document_panzoom(
+            media_url=source_url,
+            filename=active_source.filename,
+            count_label=f"Page {active_source.page_number}",
+        )
+
+
+def _render_document_source_navigation(*, document: Document, active_source: Source | None) -> None:
+    ordered = sorted(document.sources, key=lambda source: (source.page_number, source.id))
+    if not ordered or active_source is None:
+        with ui.row().classes("w-full justify-between items-center"):
+            ui.button("Previous Page", icon="chevron_left").props("flat dense disable")
+            ui.button("Next Page", icon="chevron_right").props("flat dense icon-right disable")
+        return
+    active_index = next((index for index, source in enumerate(ordered) if source.id == active_source.id), 0)
+    previous_source = ordered[active_index - 1] if active_index > 0 else None
+    next_source = ordered[active_index + 1] if active_index < len(ordered) - 1 else None
+    previous_target = f"/documents/{document.id}?source_id={previous_source.id}" if previous_source is not None else "#"
+    next_target = f"/documents/{document.id}?source_id={next_source.id}" if next_source is not None else "#"
+    with ui.row().classes("w-full justify-between items-center"):
+        previous = ui.button(
+            "Previous Page",
+            on_click=lambda: ui.navigate.to(previous_target),
+            icon="chevron_left",
+        ).props("flat dense")
+        if previous_source is None:
+            previous.props("disable")
+        following = ui.button(
+            "Next Page",
+            on_click=lambda: ui.navigate.to(next_target),
+            icon="chevron_right",
+        ).props("flat dense icon-right")
+        if next_source is None:
+            following.props("disable")
+
+
+def _render_document_detail_revision_zone(*, source: Source | None, sources_service: SourceService) -> None:
+    with ui.column().classes("col-span-12 lg:col-span-4 gap-4"), archival_card(title="Editable Revision"):
+        if source is None:
+            render_empty_state("No source pages are linked yet.", italic=True)
+            return
+        seed_revision = source.revised_text if source.revised_text is not None else (source.raw_transcription or "")
+        revision_input = (
+            ui.textarea(
+                label="Revised transcription",
+                value=seed_revision,
+            )
+            .props("outlined autogrow")
+            .classes("w-full ui-form-surface")
+        )
+        save_state = ui.label(
+            f"Last saved: {source.date_revised.isoformat()}"
+            if source.date_revised is not None
+            else "No revision saved yet."
+        ).classes("text-xs ui-text-muted")
+
+        async def submit_revision() -> None:
+            revised_text = (revision_input.value or "").strip()
+            if not revised_text:
+                ui.notify("Revised transcription cannot be empty.", type="warning")
+                return
+            save_outcome = await run_ui_action(
+                operation="documents.revision.save",
+                title="Save failed",
+                action=lambda: sources_service.upsert_revision_for_source(source_id=source.id, text=revised_text),
+            )
+            if not save_outcome.ok or save_outcome.value is None:
+                return
+            updated = save_outcome.value
+            source.revised_text = updated.revised_text
+            source.date_revised = updated.date_revised
+            save_state.text = (
+                f"Last saved: {updated.date_revised.isoformat()}"
+                if updated.date_revised is not None
+                else "Revision saved."
+            )
+            ui.notify("Revision saved", type="positive")
+
+        with ui.row().classes("w-full items-center gap-2 mt-2"):
+            ui.button("Save revision", on_click=submit_revision, icon="save").classes("ui-btn-primary")
+            ui.button(
+                "Reset",
+                on_click=lambda: _reset_document_revision_text(revision_input, source),
+                icon="refresh",
+            ).props("flat")
+
+
+def _first_source_path(document: Document) -> str | None:
+    first_source = _resolve_active_source(document, None)
+    return first_source.file_path if first_source is not None else None
 
 
 def _render_bento_metadata_zone(document: Document) -> None:
     author_names = _author_names(document)
 
-    with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
+    with ui.column().classes("w-full gap-4"):
         with archival_card(title="Archival Metadata"):
             metadata_row("Author(s):", ", ".join(author_names) if author_names else "Not set")
             metadata_row(
@@ -625,17 +775,24 @@ def _render_related_people_card(document: Document) -> None:
 
 
 def _render_document_processing_card(document: Document) -> None:
-    with archival_card(title="Sources & Pipeline Jobs"):
+    with archival_card(title="Source Pages & Transcriptions"):
-        metadata_row("Sources:", str(len(document.sources)))
+        metadata_row("Source pages:", str(len(document.sources)))
-        metadata_row("Jobs:", str(len(document.jobs)))
+        metadata_row("Transcription Jobs:", str(len(document.jobs)))
         with ui.row().classes("w-full gap-2 mt-2 flex-wrap"):
+            first_source = _resolve_active_source(document, None)
             ui.button(
-                "View Sources",
+                "View Source Detail",
-                on_click=lambda: ui.navigate.to(f"/sources?document_id={document.id}"),
+                on_click=(
+                    (lambda: ui.navigate.to(f"/sources/{first_source.id}"))
+                    if first_source is not None
+                    else (lambda: ui.notify("No source pages are linked yet.", type="warning"))
+                ),
                 icon="description",
             ).props("flat dense text-xs").classes("ui-link-primary")
             ui.button(
-                "View Jobs", on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"), icon="work_history"
+                "View Transcription Jobs",
+                on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"),
+                icon="work_history",
             ).props("flat dense text-xs").classes("ui-link-primary")
             ui.button(
                 "+ Add Job", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add"
@@ -705,3 +862,14 @@ def _resolve_selected_tag_labels(value: object) -> list[str]:
 
     labels = [candidate.strip() for candidate in flatten(value) if candidate and candidate.strip()]
     return list(dict.fromkeys(labels))
+
+
+def _latest_job_status(document: Document) -> str | None:
+    if not document.jobs:
+        return None
+    latest = max(document.jobs, key=lambda job: (job.date_created, str(job.id)))
+    return latest.status.value
+
+
+def _reset_document_revision_text(revision_input: ui.textarea, source: Source) -> None:
+    revision_input.value = source.revised_text if source.revised_text is not None else (source.raw_transcription or "")

src/transcription/ui/pages/people_page.py

@@ -789,6 +789,7 @@ def _render_linked_documents(person: Person) -> None:
             {
                 "id": str(link.document.id),
                 "document_name": link.document.name,
+                "document_date": compact_date(link.document.document_date, link.document.document_date_raw),
                 "role": link.role_ref.label if link.role_ref is not None else "Unknown role",
                 "page_count": len(link.document.sources),
             }
@@ -814,6 +815,14 @@ def _render_linked_documents(person: Person) -> None:
                 "classes": "text-left ui-table-cell-wrap",
                 "align": "left",
             },
+            {
+                "name": "document_date",
+                "label": "Document Date",
+                "field": "document_date",
+                "sortable": True,
+                "classes": "font-mono",
+                "align": "center",
+            },
             {
                 "name": "role",
                 "label": "Role",

src/transcription/ui/pages/settings_page.py

@@ -2,14 +2,19 @@
 
 from __future__ import annotations
 
+from datetime import UTC
+from datetime import datetime
 from typing import Any
 from uuid import UUID
 
+from fastapi import Request
 from nicegui import ui
 
 from transcription.config import Settings
+from transcription.db.models import MaintenanceJobType
 from transcription.runtime_helpers import run_blocking
 from transcription.services.documents import DocumentService
+from transcription.services.maintenance import MaintenanceService
 from transcription.services.people import PeopleService
 from transcription.services.prompts import PromptStore
 from transcription.ui.components.app_shell import render_navigation_header
@@ -25,6 +30,7 @@ from transcription.ui.runtime_settings_store import HIDDEN_SETTINGS_CATEGORIES
 from transcription.ui.runtime_settings_store import read_runtime_settings_snapshot
 from transcription.ui.runtime_settings_store import save_runtime_settings
 from transcription.ui.theme import page_header
+from transcription.worker import resolve_worker_notifier
 
 from ...db.session import SessionFactoryDep
 
@@ -33,9 +39,10 @@ def register_page(*, settings: Settings) -> None:  # noqa: PLR0915
     """Register the constrained Settings route."""
 
     @ui.page("/settings")
-    async def settings_page(session_factory: SessionFactoryDep) -> None:  # noqa: PLR0915
+    async def settings_page(request: Request, session_factory: SessionFactoryDep) -> None:  # noqa: PLR0915
         documents = DocumentService(session_factory=session_factory)
         people = PeopleService(session_factory=session_factory)
+        maintenance = MaintenanceService(session_factory=session_factory)
         prompts = PromptStore(settings=settings)
         render_navigation_header(current_path="/settings")
 
@@ -500,6 +507,158 @@ def register_page(*, settings: Settings) -> None:  # noqa: PLR0915
                     with ui.row().classes("items-center gap-2"):
                         ui.button("Save home text", icon="save", on_click=save_home_text).classes("ui-btn-primary")
 
+            @ui.refreshable
+            async def render_maintenance() -> None:
+                with archival_card("Maintenance"):
+                    ui.label(
+                        "Queue maintenance tasks for worker execution. Runs are persisted with summary and logs."
+                    ).classes("text-xs ui-text-muted mb-3")
+                    with ui.row().classes("w-full items-center gap-2"):
+                        ui.button(
+                            "Run Backup",
+                            icon="save",
+                            on_click=lambda: _enqueue_maintenance_run(
+                                maintenance=maintenance,
+                                job_type=MaintenanceJobType.BACKUP,
+                                request=request,
+                                refresh=render_maintenance.refresh,
+                            ),
+                        ).classes("ui-btn-primary")
+                        ui.button(
+                            "Run Storage Reconciliation",
+                            icon="rule",
+                            on_click=lambda: _enqueue_maintenance_run(
+                                maintenance=maintenance,
+                                job_type=MaintenanceJobType.STORAGE_RECONCILIATION,
+                                request=request,
+                                refresh=render_maintenance.refresh,
+                            ),
+                        ).props("flat")
+
+                    runs_outcome = await run_ui_action(
+                        operation="settings.maintenance.list",
+                        title="Maintenance runs unavailable",
+                        action=maintenance.list_runs,
+                    )
+                    if not runs_outcome.ok:
+                        return
+                    runs = list(runs_outcome.value or ())
+                    if not runs:
+                        render_empty_state("No maintenance runs recorded yet.", extra_classes="mt-3")
+                        return
+
+                    rows = [
+                        {
+                            "id": str(run.id),
+                            "job_type": run.job_type.value.replace("_", " ").title(),
+                            "status": run.status.value,
+                            "started_at": _format_timestamp(run.started_at),
+                            "finished_at": _format_timestamp(run.finished_at),
+                            "duration": _format_duration(started_at=run.started_at, finished_at=run.finished_at),
+                            "summary": run.summary or "-",
+                            "log_path": run.log_path or "",
+                            "error_detail": run.error_detail or "",
+                        }
+                        for run in runs
+                    ]
+
+                    table = (
+                        ui.table(
+                            rows=rows,
+                            columns=[
+                                {"name": "job_type", "label": "Job Type", "field": "job_type", "sortable": True},
+                                {"name": "status", "label": "Status", "field": "status", "sortable": True},
+                                {"name": "started_at", "label": "Started", "field": "started_at", "sortable": True},
+                                {"name": "finished_at", "label": "Finished", "field": "finished_at", "sortable": True},
+                                {"name": "duration", "label": "Duration", "field": "duration", "sortable": False},
+                                {"name": "summary", "label": "Summary", "field": "summary", "sortable": False},
+                            ],
+                            row_key="id",
+                            selection="single",
+                        )
+                        .classes("w-full ui-table")
+                        .props(
+                            'flat square table-style="table-layout: fixed; width: 100%;" '
+                            'header-cell-class="ui-table-header text-xs uppercase tracking-wider" '
+                            'table-class="ui-table-body text-xs"'
+                        )
+                    )
+
+                    table.add_slot(
+                        "body-cell-status",
+                        r"""
+                        <q-td :props="props">
+                            <q-chip
+                                dense
+                                square
+                                size="sm"
+                                :class="`ui-status ui-status--${props.value}`"
+                            >
+                                {{ props.value.toUpperCase() }}
+                            </q-chip>
+                        </q-td>
+                        """,
+                    )
+                    run_by_id = {str(run.id): run for run in runs}
+
+                    def selected_run_id() -> str | None:
+                        selected = table.selected or []
+                        if len(selected) != 1:
+                            return None
+                        return str(selected[0].get("id") or "")
+
+                    async def view_log() -> None:
+                        run_id = selected_run_id()
+                        if not run_id:
+                            ui.notify("Select one run first.", type="warning")
+                            return
+                        run = run_by_id.get(run_id)
+                        if run is None or not run.log_path:
+                            ui.notify("Log unavailable for this run.", type="warning")
+                            return
+                        log_path = run.log_path
+                        log_outcome = await run_ui_action(
+                            operation="settings.maintenance.log.read",
+                            title="Maintenance log unavailable",
+                            action=lambda: _read_maintenance_log(maintenance=maintenance, log_path=log_path),
+                        )
+                        if not log_outcome.ok or log_outcome.value is None:
+                            return
+                        with ui.dialog() as dialog, ui.card().classes("w-full max-w-4xl"):
+                            ui.label(f"Log: {run.log_path}").classes("text-sm font-semibold")
+                            ui.code(log_outcome.value.decode("utf-8", errors="replace"), language="text").classes(
+                                "w-full text-xs max-h-[65vh] overflow-auto"
+                            )
+                            with ui.row().classes("w-full justify-end"):
+                                ui.button("Close", on_click=dialog.close).props("flat")
+                        dialog.open()
+
+                    async def download_log() -> None:
+                        run_id = selected_run_id()
+                        if not run_id:
+                            ui.notify("Select one run first.", type="warning")
+                            return
+                        run = run_by_id.get(run_id)
+                        if run is None or not run.log_path:
+                            ui.notify("Log unavailable for this run.", type="warning")
+                            return
+                        log_path = run.log_path
+                        log_outcome = await run_ui_action(
+                            operation="settings.maintenance.log.download",
+                            title="Maintenance log unavailable",
+                            action=lambda: _read_maintenance_log(maintenance=maintenance, log_path=log_path),
+                        )
+                        if not log_outcome.ok or log_outcome.value is None:
+                            return
+                        ui.download(log_outcome.value, filename=f"{run.id}.log", media_type="text/plain")
+
+                    with ui.row().classes("w-full justify-end items-center gap-2 mt-2"):
+                        ui.button("View Log", icon="visibility", on_click=view_log).props("flat")
+                        ui.button("Download Log", icon="download", on_click=download_log).props("flat")
+
+                    if any(run.status.value in {"queued", "processing"} for run in runs):
+                        ui.timer(4.0, render_maintenance.refresh, once=True)
+
             @ui.refreshable
             async def render_runtime_settings() -> None:
                 with archival_card("Runtime Settings"):
@@ -593,6 +752,7 @@ def register_page(*, settings: Settings) -> None:  # noqa: PLR0915
                 tags_tab = ui.tab("Tags")
                 prompts_tab = ui.tab("Prompts")
                 home_page_text_tab = ui.tab("Home Page Text")
+                maintenance_tab = ui.tab("Maintenance")
                 runtime_settings_tab = ui.tab("Runtime Settings")
 
             with ui.tab_panels(tabs, value=document_types_tab).classes("w-full"):
@@ -606,6 +766,8 @@ def register_page(*, settings: Settings) -> None:  # noqa: PLR0915
                     await render_prompts()
                 with ui.tab_panel(home_page_text_tab):
                     await render_home_page_text()
+                with ui.tab_panel(maintenance_tab):
+                    await render_maintenance()
                 with ui.tab_panel(runtime_settings_tab):
                     await render_runtime_settings()
 
@@ -647,3 +809,48 @@ async def _read_runtime_settings_snapshot(settings: Settings):
 
 async def _write_runtime_settings(*, settings: Settings, updates: dict[str, str | bool]):
     return await run_blocking(save_runtime_settings, settings=settings, updates=updates)
+
+
+async def _enqueue_maintenance_run(
+    *,
+    maintenance: MaintenanceService,
+    job_type: MaintenanceJobType,
+    request: Request,
+    refresh,
+) -> None:
+    created_outcome = await run_ui_action(
+        operation="settings.maintenance.enqueue",
+        title="Maintenance run failed",
+        action=lambda: maintenance.enqueue_run(job_type=job_type),
+    )
+    if not created_outcome.ok or created_outcome.value is None:
+        return
+    resolve_worker_notifier(request.app.state).notify()
+    ui.notify(f"Queued {job_type.value.replace('_', ' ')} run", type="positive")
+    refresh()
+
+
+def _format_timestamp(value: datetime | None) -> str:
+    if value is None:
+        return "-"
+    parsed = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+    return parsed.astimezone().strftime("%b %d, %I:%M %p")
+
+
+def _format_duration(*, started_at: datetime | None, finished_at: datetime | None) -> str:
+    if started_at is None:
+        return "-"
+    if finished_at is None:
+        return "in progress"
+    elapsed = finished_at - started_at
+    seconds = int(elapsed.total_seconds())
+    if seconds < 1:
+        return "<1s"
+    minutes, remainder = divmod(seconds, 60)
+    if minutes:
+        return f"{minutes}m {remainder}s"
+    return f"{remainder}s"
+
+
+async def _read_maintenance_log(*, maintenance: MaintenanceService, log_path: str) -> bytes:
+    return await run_blocking(maintenance.read_log_bytes, log_path=log_path)

src/transcription/ui/pages/sources_page.py

@@ -9,6 +9,7 @@ from pathlib import Path
 from uuid import UUID
 
 from fastapi import Request
+from fastapi.responses import RedirectResponse
 from nicegui import ui
 from PIL import Image
 from PIL import UnidentifiedImageError
@@ -62,12 +63,15 @@ def register_page() -> None:  # noqa: PLR0915
         session_factory: SessionFactoryDep,
         document_id: str | None = None,
         job_id: str | None = None,
-    ) -> None:
+    ) -> RedirectResponse | None:
         sources_service = SourceService(session_factory=session_factory)
         parsed_doc_id = parse_uuid(document_id)
         parsed_job_id = parse_uuid(job_id)
 
-        header_title = "Source Asset Records"
+        if parsed_doc_id is None and parsed_job_id is None:
+            return RedirectResponse(url="/ui/documents")
+
+        header_title = ""
         if parsed_doc_id is not None:
             header_title = "Sources for Document"
         elif parsed_job_id is not None:
@@ -101,10 +105,10 @@ def register_page() -> None:  # noqa: PLR0915
                     ).props("flat")
                 else:
                     ui.button(
-                        "Create Job",
+                        "Back to Documents",
-                        on_click=lambda: ui.navigate.to("/jobs/new"),
+                        on_click=lambda: ui.navigate.to("/documents"),
-                        icon="add",
+                        icon="arrow_back",
-                    ).classes("ui-btn-primary")
+                    ).props("flat")
 
             rows = [
                 SourceTableRow(
@@ -120,11 +124,6 @@ def register_page() -> None:  # noqa: PLR0915
             ]
             render_sources_table(rows)
 
-            if parsed_doc_id is None and parsed_job_id is None:
-                ui.label("Open a source row to inspect AI output and add human revisions.").classes(
-                    "text-xs ui-text-muted"
-                )
-
     @ui.page("/sources/{source_id}")
     async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
         sources_service = SourceService(session_factory=session_factory)
@@ -164,8 +163,8 @@ def register_page() -> None:  # noqa: PLR0915
                 )
                 with ui.row().classes("items-center gap-2"):
                     ui.button(
-                        "Back to Sources",
+                        "Back to Document",
-                        on_click=lambda: ui.navigate.to("/sources"),
+                        on_click=lambda: ui.navigate.to(f"/documents/{source.document_id}?source_id={source.id}"),
                         icon="arrow_back",
                     ).props("flat")
                     ui.button(
@@ -267,14 +266,14 @@ def register_page() -> None:  # noqa: PLR0915
                     return
                 except TranscriptionNotFoundError:
                     ui.notify("Source not found.", type="warning")
-                    ui.navigate.to("/sources")
+                    ui.navigate.to("/documents")
                     return
                 except Exception as exc:  # noqa: BLE001
                     show_error(exc, title="Delete failed", operation="sources.delete")
                     return
 
                 ui.notify("Source deleted", type="positive")
-                ui.navigate.to("/sources")
+                ui.navigate.to("/documents")
 
             render_delete_actions(
                 confirm_label="Delete source permanently",

src/transcription/ui/static/theme.css

@@ -299,7 +299,8 @@ input:focus-visible,
 }
 
 .ui-status--partial_success,
-.ui-status--transcribed {
+.ui-status--transcribed,
+.ui-status--succeeded {
   color: var(--theme-text);
   background: var(--theme-secondary);
 }

src/transcription/worker.py

@@ -229,17 +229,20 @@ async def run_worker_loop(
 
             processed_any = False
             while True:
+                processed_job = False
                 with handle_worker_exceptions(operation="worker.process_next_queued_job"):
-                    processed = await process_next_queued_job(
+                    processed_job = await process_next_queued_job(
                         session_factory=session_factory,
                         services=services,
                     )
-                    if not processed:
+                processed_maintenance = False
-                        break
+                if session_factory is not None:
-                    processed_any = True
+                    with handle_worker_exceptions(operation="worker.process_next_queued_maintenance"):
-                    continue
+                        processed_maintenance = await services.maintenance.process_next_queued_run()
 
-                break
+                if not processed_job and not processed_maintenance:
+                    break
+                processed_any = True
 
             if wake_event is None and not processed_any:
                 await asyncio.sleep(poll_interval_seconds)

tests/services/test_maintenance_service.py

@@ -0,0 +1,59 @@
+"""Tests for maintenance run persistence and execution lifecycle."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from transcription.config import Settings
+from transcription.db.models import MaintenanceJobType
+from transcription.db.models import MaintenanceRunStatus
+from transcription.services.maintenance import MaintenanceExecution
+from transcription.services.maintenance import MaintenanceService
+
+
+@pytest.mark.asyncio
+async def test_enqueue_and_list_runs(default_session_factory, default_settings):
+    service = MaintenanceService(session_factory=default_session_factory, settings=default_settings)
+
+    first = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
+    second = await service.enqueue_run(job_type=MaintenanceJobType.STORAGE_RECONCILIATION, triggered_by="test")
+    runs = await service.list_runs(limit=10)
+
+    assert len(runs) == 2
+    assert runs[0].id == second.id
+    assert runs[1].id == first.id
+    assert runs[0].status == MaintenanceRunStatus.QUEUED
+
+
+@pytest.mark.asyncio
+async def test_process_next_queued_run_persists_terminal_result(
+    default_session_factory,
+    default_settings,
+    tmp_path,
+    monkeypatch,
+):
+    settings = default_settings.model_copy(update={"log_dir": tmp_path / "logs"})
+    settings = Settings.model_validate(settings.model_dump())
+    service = MaintenanceService(session_factory=default_session_factory, settings=settings)
+    queued = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test")
+
+    async def _fake_execute(_run):
+        return MaintenanceExecution(
+            status=MaintenanceRunStatus.SUCCEEDED,
+            summary="Synthetic success",
+            output="stdout line\nstderr line",
+        )
+
+    monkeypatch.setattr(service, "_execute_run", _fake_execute)
+
+    processed = await service.process_next_queued_run()
+    assert processed is True
+
+    runs = await service.list_runs(limit=10)
+    updated = next(run for run in runs if run.id == queued.id)
+    assert updated.status == MaintenanceRunStatus.SUCCEEDED
+    assert updated.summary == "Synthetic success"
+    assert updated.log_path is not None
+    assert (settings.log_dir / Path(updated.log_path)).is_file()

tests/test_worker.py

@@ -204,3 +204,54 @@ async def test_process_next_leaves_a_caller_owned_bundle_open(monkeypatch):
 
     assert await process_next_queued_job(services=bundle) is False
     assert closed is False
+
+
+@pytest.mark.asyncio
+async def test_run_worker_loop_processes_queued_maintenance_runs(monkeypatch):
+    stop_event = asyncio.Event()
+    maintenance_calls = 0
+
+    class _Sources:
+        async def aclose(self):
+            return
+
+    class _Maintenance:
+        async def process_next_queued_run(self):
+            nonlocal maintenance_calls
+            maintenance_calls += 1
+            if maintenance_calls == 1:
+                return True
+            stop_event.set()
+            return False
+
+    class _Jobs:
+        settings = Settings(openrouter_api_key="test-key", worker_stale_job_seconds=120.0)
+
+        async def requeue_stale_processing_jobs(self, *, stale_before, session=None):
+            _ = (stale_before, session)
+            return 0
+
+    class _Bundle:
+        def __init__(self):
+            self.jobs = _Jobs()
+            self.sources = cast("SourceService", _Sources())
+            self.maintenance = _Maintenance()
+
+        async def aclose(self):
+            await self.sources.aclose()
+
+    monkeypatch.setattr(
+        "transcription.worker.ServiceBundle.from_session_factory",
+        classmethod(lambda _cls, _factory=None, **_kwargs: cast(ServiceBundle, _Bundle())),
+    )
+
+    async def _no_jobs(*, session=None, session_factory=None, services=None):
+        _ = (session, session_factory, services)
+        return False
+
+    monkeypatch.setattr("transcription.worker.process_next_queued_job", _no_jobs)
+
+    fake_session_factory = cast(async_sessionmaker[AsyncSession], object())
+    await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0, session_factory=fake_session_factory)
+
+    assert maintenance_calls >= 2

tests/ui/test_documents_page.py

@@ -103,7 +103,8 @@ class TestDocumentsPageRendering:
             r'"name":"tags","label":"Tags".*'
             r'"name":"document_date","label":"Document Date".*'
             r'"name":"document_type","label":"Type".*'
-            r'"name":"source_count","label":"# Sources"',
+            r'"name":"source_count","label":"# Sources".*'
+            r'"name":"transcription_status","label":"Transcription Status"',
             response.text,
             re.DOTALL,
         )
@@ -181,17 +182,23 @@ class TestDocumentsPageRendering:
 
         assert response.status_code == 200
         assert "Letter from Hig" in response.text
-        assert "ZC-1924-001" in response.text
         assert "Zenna Cochran" in response.text
-        assert "Document Type:" in response.text
+        assert "Document Details" in response.text
-        assert "Letter" in response.text
-        assert "07-04-1924" in response.text
-        assert "1924-07-04" not in response.text
-        assert "google.com/maps/search/?api=1&query=Salt+Lake+City%2C+Utah" in response.text
-        assert "PIPELINE JOBS" in response.text.upper()
         assert "Edit Document" in response.text
         assert "Back to Documents" in response.text
 
+    @pytest.mark.asyncio
+    async def test_document_info_page_renders_metadata_cards(self, app_client, seed_person_and_document):
+        _, client = app_client
+        doc_id, _ = seed_person_and_document
+
+        response = client.get(f"/ui/documents/{doc_id}/info")
+
+        assert response.status_code == 200
+        assert "Document Info" in response.text
+        assert "ZC-1924-001" in response.text
+        assert "google.com/maps/search/?api=1&query=Salt+Lake+City%2C+Utah" in response.text
+
     @pytest.mark.asyncio
     async def test_document_detail_page_renders_person_context_back_button(self, app_client, seed_person_and_document):
         _, client = app_client

tests/ui/test_navigation_and_mounts.py

@@ -28,9 +28,7 @@ class TestNavigationAndMounts:
             "/ui/homepage",
             "/ui/homepage/edit",
             "/ui/documents",
-            "/ui/tags",
             "/ui/people",
-            "/ui/sources",
             "/ui/jobs",
             "/ui/settings",
         ],

tests/ui/test_pages_registration.py

@@ -14,17 +14,13 @@ class TestPageRegistration:
         homepage_response = client.get("/ui/homepage")
         documents_response = client.get("/ui/documents")
         people_response = client.get("/ui/people")
-        sources_response = client.get("/ui/sources")
         jobs_response = client.get("/ui/jobs")
-        tags_response = client.get("/ui/tags")
         settings_response = client.get("/ui/settings")
 
         assert homepage_response.status_code == 200
         assert documents_response.status_code == 200
         assert people_response.status_code == 200
-        assert sources_response.status_code == 200
         assert jobs_response.status_code == 200
-        assert tags_response.status_code == 200
         assert settings_response.status_code == 200
         assert "Runtime Settings" in settings_response.text
         assert "Document Types" in settings_response.text
@@ -32,5 +28,6 @@ class TestPageRegistration:
         assert "Tags" in settings_response.text
         assert "Prompts" in settings_response.text
         assert "Home Page Text" in settings_response.text
+        assert "Maintenance" in settings_response.text
         assert "Other settings not shown here" in settings_response.text
         assert "README.md" not in settings_response.text

tests/ui/test_people_page.py

@@ -306,6 +306,7 @@ class TestPeoplePageRendering:
 
         assert response.status_code == 200
         assert "Document Name" in response.text
+        assert "Document Date" in response.text
         assert "Number of Pages" in response.text
         assert "Linked Document" in response.text
         assert "Author" in response.text

tests/ui/test_sources_page.py

@@ -165,12 +165,10 @@ class TestSourcesPageRendering:
     def test_sources_page_renders_empty_state(self, app_client):
         _, client = app_client
 
-        response = client.get("/ui/sources")
+        response = client.get("/ui/sources", follow_redirects=False)
 
-        assert response.status_code == 200
+        assert response.status_code == 307
-        assert "Source Asset Records" in response.text
+        assert response.headers["location"] == "/ui/documents"
-        assert "No source asset records found in repository." in response.text
-        assert "Upload New Documents" not in response.text
 
     @pytest.mark.asyncio
     async def test_sources_page_lists_seeded_sources(self, app_client):
@@ -192,10 +190,12 @@ class TestSourcesPageRendering:
                 )
             )
             await session.commit()
+            document_id = str(document.id)
 
-        response = client.get("/ui/sources")
+        response = client.get(f"/ui/sources?document_id={document_id}")
 
         assert response.status_code == 200
+        assert "Sources for Document" in response.text
         assert "page_one.png" in response.text
         assert "Source Document" in response.text
         assert "Stored Filename" not in response.text

tests/ui/test_tags_page.py

@@ -1,43 +1,13 @@
-"""Tests for the tags page route and grouped filtering behavior."""
+"""Tags route retirement guards."""
 
 import pytest
 
-from transcription.db import session_scope
-from transcription.db.models import Document
-from transcription.services.documents import DocumentService
-
 
 @pytest.mark.integration
-class TestTagsPageRendering:
+class TestTagsPageRetired:
-    def test_tags_page_renders_empty_state_without_tags(self, app_client):
+    def test_tags_page_route_is_not_registered(self, app_client):
         _, client = app_client
 
         response = client.get("/ui/tags")
 
-        assert response.status_code == 200
+        assert response.status_code == 404
-        assert "Tags" in response.text
-        assert "No tags are configured yet." in response.text
-
-    @pytest.mark.asyncio
-    async def test_tags_page_groups_documents_by_tag(self, app_client):
-        app, client = app_client
-        documents = DocumentService(session_factory=app.state.runtime.session_factory)
-
-        async with session_scope(session_factory=app.state.runtime.session_factory) as session:
-            first = Document(name="Tagged Letter")
-            second = Document(name="Tagged Journal")
-            session.add_all([first, second])
-            await session.flush()
-            await documents.sync_document_tags_by_labels(document_id=first.id, labels=["Family"], session=session)
-            await documents.sync_document_tags_by_labels(
-                document_id=second.id,
-                labels=["Family", "Research"],
-                session=session,
-            )
-            await session.commit()
-
-        response = client.get("/ui/tags")
-
-        assert response.status_code == 200
-        assert "Filter by tag" in response.text
-        assert "Tags" in response.text