Merge branch 'main' into diverging

docs/index.md

@@ -6,7 +6,7 @@ This project is a production application for transcribing and preserving histori
 
 Read [architecture.md](architecture.md) first.
 
-Then review [ver1/ver1.md](ver1/ver1.md) for completion scope and [ver1/ver1-step1-results.md](ver1/ver1-step1-results.md) for current architecture-consolidation status.
+Then review [ver1/ver1.md](ver1/ver1.md) for completion scope.
 
 The architecture page is the primary technical reference and defines:
 
@@ -41,10 +41,8 @@ This operating model keeps deployment and maintenance simple while preserving cl
 
 ## Documentation Map
 
-- Architecture and technical design: [architecture.md](architecture.md)
 - Version 1 implementation plan: [ver1/ver1.md](ver1/ver1.md)
-- Version 1 Step 1 plan: [ver1/ver1-step1.md](ver1/ver1-step1.md)
+- Architecture and technical design: [architecture.md](architecture.md)
-- Version 1 Step 1 results: [ver1/ver1-step1-results.md](ver1/ver1-step1-results.md)
 - Architecture decision records (ADR index): [adr/README.md](adr/README.md)
 - Runtime and deployment requirements: [requirements.md](requirements.md)
 - Error handling policy and operational guidance: [error_handling.md](error_handling.md)

docs/ver1/ver1-step6.md

@@ -0,0 +1,206 @@
+## Step 6 Goal (from `docs/ver1/ver1.md`)
+Implement **minimal observability & operability** so a single operator can quickly diagnose and recover from common failures.
+
+---
+
+## 1) Current-State Assessment (what already exists)
+
+### Already in place
+- Central startup logging initialization via `setup_logging()` and `dictConfig` (`src/transcription/config.py`, `src/transcription/app.py`).
+- Error taxonomy and `error_id` envelope contract (`src/transcription/errors.py`) aligned with `docs/error_handling.md`.
+- Error handling for API and worker includes category + error IDs in some paths (`src/transcription/api/errors.py`, `src/transcription/worker.py`).
+- Basic health endpoint `/healthz` (`src/transcription/api/health.py`).
+- UI error display already shows actionable message + error reference (`src/transcription/ui/error_presenter.py`).
+
+### Gaps to close for Step 6
+1. **Structured logging is inconsistent** (many logs are free-form text with embedded key/value; no enforced schema).
+2. **Boundary coverage is incomplete** (UI/service/API/worker don’t all emit consistent operation logs).
+3. `/healthz` is very basic; no lightweight readiness/startup diagnostics endpoint/reporting.
+4. No concise **operator runbook** yet (start/stop, log interpretation, recovery playbooks).
+5. Minimal counters/timings are not yet standardized.
+
+---
+
+## 2) MCP Guidance Incorporated (relevant items)
+
+From `john-stream-mcp`, these are directly applied:
+
+- **`python-logging-dictconfig`**: keep one centralized `dictConfig`, configure once at startup, named loggers in modules.
+- **`fastapi-async-sqlalchemy-modernization`**: include observability + health/readiness checks; explicit lifecycle and deterministic startup/shutdown checks.
+- **`fastapi-uv-docker`**: keep `/healthz`; add practical readiness/ops checks for deployment clarity.
+- **`pytesting`**: deterministic tests, concise structure, validation lanes (`collect-only`, `unit`, `not external`, full).
+- **`pydantic-settings`**: keep typed settings as single source for logging/health behavior flags.
+- **`nicegui` + `nicegui-ui-customization`**: preserve clear, actionable user-facing error feedback and non-blocking UI flows.
+- **`zensical-docs`**: produce focused, navigable operator docs.
+
+(Other MCP resources were reviewed but are not core to Step 6 implementation scope.)
+
+---
+
+## 3) Detailed Implementation Plan for Step 6
+
+## Workstream A — Structured Logging Contract
+
+### A1. Define a canonical log event schema
+Create a project log schema (doc + code-level constants) with required keys:
+- `timestamp` (UTC)
+- `level`
+- `logger`
+- `operation`
+- `event`
+- `error_id` (when error)
+- `category` (when error)
+- `exception_type` (when error)
+- `job_id`, `document_id` (when relevant)
+- optional: `duration_ms`, `retry_count`, `status`
+
+### A2. Standardize log emission helpers
+Add small logging helpers (or adapter utilities) to reduce drift:
+- `log_operation_start(...)`
+- `log_operation_success(...)`
+- `log_operation_error(...)`
+
+Keep this minimal and avoid heavy observability frameworks.
+
+### A3. Update formatter to structured output
+Use `dictConfig` to emit either:
+- JSON lines (preferred for structure), or
+- strict key-value line format with fixed fields.
+
+**Recommendation:** JSON lines to satisfy “structured logging” unambiguously while still simple.
+
+---
+
+## Workstream B — Boundary-by-Boundary Instrumentation
+
+### B1. API boundary (`src/transcription/api/*`)
+- Add request-level operation logs for key routes (`upload.submit`, `jobs.list`, `jobs.get`, etc.).
+- Ensure API exception handler logs always include `error_id`, `category`, `operation`, `exception_type`.
+
+### B2. Service boundary (`src/transcription/services/*`)
+- Add operation logs around:
+  - upload validation/persist,
+  - transcription orchestration,
+  - revision add/accept,
+  - search/export.
+- Add timing (`duration_ms`) for high-value operations only.
+
+### B3. Worker boundary (`src/transcription/worker.py`)
+- Standardize all worker log events to schema.
+- Ensure retry logs include: `retriable`, `retry_count`, `max_retries`, `backoff_seconds`.
+- Ensure terminal failure logs include error contract fields.
+
+### B4. UI boundary (`src/transcription/ui/*`)
+- Keep user-safe UI messages as-is.
+- Add backend/UI logger events for user-triggered failures (operation + error_id + category) so UI-visible errors correlate to server logs.
+
+---
+
+## Workstream C — Health, Readiness, Startup Operability
+
+### C1. Keep `/healthz` lightweight
+- Return “process is running” status quickly.
+
+### C2. Add lightweight `/readyz`
+Include small checks:
+- DB connectivity ping.
+- Worker thread alive check.
+- Optional prompt directory existence check.
+
+Return structured status payload with per-check pass/fail.
+
+### C3. Startup self-check summary log
+At startup, emit one concise ops summary event:
+- environment
+- schema validation result
+- worker started
+- directories checked
+- bootstrap/migration mode flags
+
+---
+
+## Workstream D — Minimal Counters & Timings
+
+Add only high-value diagnostics:
+1. `worker_jobs_processed_total`
+2. `worker_jobs_failed_total`
+3. `worker_retries_total`
+4. `transcription_duration_ms` (per job)
+5. `upload_persist_duration_ms` (per upload path)
+
+Implementation can be log-derived counters (no external metrics backend required).
+
+---
+
+## Workstream E — Operator Runbook
+
+Create concise runbook doc (recommended: `docs/ver1/ver1-step6-operator-runbook.md`) with:
+
+1. **Start/Stop**
+   - local `uv` run mode
+   - docker compose mode (if applicable)
+
+2. **Where logs are**
+   - stdout, docker logs commands, filtering by `error_id` / `operation`.
+
+3. **Common failure patterns → recovery**
+   - provider timeout
+   - auth denied
+   - missing prompt dir
+   - DB unavailable
+   - job stuck/failed with retry exhausted
+
+4. **Recovery procedures**
+   - restart sequence
+   - verify health/readiness
+   - when to requeue/re-upload
+
+5. **Escalation artifacts**
+   - capture timestamp + error_id + operation + job_id/document_id
+
+Also update `README.md` with short links to the runbook.
+
+---
+
+## Workstream F — Verification & Quality Gates
+
+### Tests to add/update
+- `tests/api/test_health.py`
+  - `/healthz` baseline
+  - `/readyz` pass/fail behavior
+- `tests/api/test_error_responses.py` / `tests/api/test_routes.py`
+  - logs include `error_id/category/operation` on failures
+- `tests/services/test_worker.py`
+  - retry/failure log fields + timing presence
+- `tests/ui/*`
+  - ensure UI error correlation path includes operation/ref id behavior
+
+### Validation commands (per MCP pytest guidance)
+- `uv run pytest --collect-only -q`
+- `uv run pytest -m unit -q`
+- `uv run pytest -m "not external" -q`
+- `uv run pytest -q`
+
+---
+
+## 4) Traceability to Governing Docs
+
+- **`docs/ver1/ver1.md` Step 6:** all 5 implementation bullets covered.
+- **`docs/error_handling.md`:** logging contract fields and error taxonomy continuity enforced.
+- **`docs/architecture.md`:** respects modular boundaries, in-process worker model, low-complexity ops.
+- **`docs/requirements.md`:**
+  - REQ-8 (startup logging/config centralization) strengthened,
+  - REQ-5 (status visibility) improved operationally,
+  - REQ-7 lifecycle ownership observability improved.
+- **`docs/intent.md`:** keeps operation simple for personal-scale archival workflow.
+
+---
+
+## 5) Suggested Execution Order (low risk)
+
+1. Logging schema + formatter + helpers  
+2. Worker/API instrumentation (highest value)  
+3. Service/UI instrumentation  
+4. `/readyz` + startup summary check  
+5. Runbook + README links  
+6. Tests + Step 6 results artifact (`docs/ver1/ver1-step6-results.md`)

src/transcription/api/errors.py

@@ -11,7 +11,6 @@ from fastapi.responses import JSONResponse
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 from transcription.errors import build_error_envelope
-from transcription.security import AccessDeniedError
 
 logger = logging.getLogger(__name__)
 

src/transcription/app.py

@@ -7,33 +7,27 @@ from threading import Event
 from threading import Thread
 
 from fastapi import FastAPI
-from fastapi import Request
+from sqlalchemy.ext.asyncio import async_sessionmaker
-from fastapi.responses import JSONResponse
+from sqlmodel.ext.asyncio.session import AsyncSession
 
-from transcription.api.errors import register_error_handlers
+from .api.errors import register_error_handlers
-from transcription.api.health import router as health_router
+from .api.health import router as health_router
-from transcription.api.routes import router as transcription_router
+from .config import configure_logging
-from transcription.config import get_settings
+from .config import get_settings
-from transcription.config import setup_logging
+from .db import cleanup_database
-from transcription.db import create_all
+from .db import create_all
-from transcription.db import dispose_database_runtime
+from .db import initialize_database_runtime
-from transcription.db import initialize_database_runtime
+from .ui import register_pages
-from transcription.db import should_bootstrap_schema
+from .worker import run_worker_loop
-from transcription.db import validate_schema_compatibility
-from transcription.errors import build_error_envelope
-from transcription.migrations import apply_pending_migrations
-from transcription.security import AccessDeniedError
-from transcription.security import enforce_request_access
-from transcription.ui import register_pages
-from transcription.worker import run_worker_loop
 
 
 def _start_worker(app: FastAPI) -> None:
+    session_factory: async_sessionmaker[AsyncSession] = app.state.db_session_factory
     stop_event = Event()
     worker_thread = Thread(
         target=run_worker_loop,
         kwargs={
-            "engine": app.state.db_runtime.engine,
+            "session_factory": session_factory,
             "stop_event": stop_event,
             "poll_interval_seconds": 1.0,
         },
@@ -56,23 +50,16 @@ def _stop_worker(app: FastAPI) -> None:
 
 @asynccontextmanager
 async def _lifespan(app: FastAPI):
-    setup_logging()
+    configure_logging()
 
     settings = get_settings()
     app.state.settings = settings
-    app.state.db_runtime = initialize_database_runtime(settings=settings)
+    runtime = initialize_database_runtime(settings=settings)
+    app.state.db_engine = runtime.engine
+    app.state.db_session_factory = runtime.session_factory
 
-    if should_bootstrap_schema(settings):
+    if settings.should_bootstrap_schema:
-        create_all(engine=app.state.db_runtime.engine)
+        await create_all(engine=runtime.engine)
-
-    if settings.migration_auto_apply_on_startup:
-        apply_pending_migrations(engine=app.state.db_runtime.engine)
-
-    if settings.validate_schema_on_startup:
-        compatibility_issues = validate_schema_compatibility(engine=app.state.db_runtime.engine)
-        if compatibility_issues:
-            issues_text = ", ".join(compatibility_issues)
-            raise RuntimeError(f"Schema compatibility check failed: {issues_text}")
 
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
@@ -82,7 +69,7 @@ async def _lifespan(app: FastAPI):
         yield
     finally:
         _stop_worker(app)
-        dispose_database_runtime()
+        await cleanup_database()
 
 
 def create_app() -> FastAPI:

src/transcription/config.py

@@ -6,15 +6,16 @@ are resolved by the provider adapters, not here.
 """
 
 import logging.config
+from contextvars import ContextVar
 from enum import StrEnum
-from functools import lru_cache
 from pathlib import Path
 from typing import Literal
 
-from pydantic import model_validator
 from pydantic_settings import BaseSettings
 from pydantic_settings import SettingsConfigDict
 
+logger = logging.getLogger(__name__)
+
 
 class Provider(StrEnum):
     OPENROUTER = "openrouter"
@@ -59,11 +60,23 @@ class Settings(BaseSettings):
     worker_max_retries: int = 0
     worker_retry_backoff_seconds: float = 0.0
 
-    @model_validator(mode="after")
+    @property
-    def _validate_operator_access_settings(self) -> "Settings":
+    def should_bootstrap_schema(self) -> bool:
-        if self.operator_access_enabled and not self.operator_password:
+        """Return whether startup should auto-create schema for this environment."""
-            raise ValueError("OPERATOR_PASSWORD is required when OPERATOR_ACCESS_ENABLED=true")
+        if self.bootstrap_schema_on_startup is not None:
-        return self
+            return self.bootstrap_schema_on_startup
+        return self.environment in {"development", "test"}
+
+
+_settings: ContextVar[Settings | None] = ContextVar("settings", default=None)
+
+
+def get_settings() -> Settings:
+    settings = _settings.get()
+    if settings is None:
+        settings = Settings()  # pyright: ignore[reportCallIssue]
+        _settings.set(settings)
+    return settings
 
 
 LOGGING_CONFIG: dict[str, object] = {
@@ -86,18 +99,17 @@ LOGGING_CONFIG: dict[str, object] = {
         "level": "INFO",
         "handlers": ["console"],
     },
+    "loggers": {
+        "transcription": {
+            "level": "DEBUG",
+            "handlers": ["console"],
+            "propagate": False,
+        }
+    },
 }
 
 
-@lru_cache(maxsize=1)
+def configure_logging() -> None:
-def get_settings() -> Settings:
-    """Return the singleton Settings instance.
-
-    Cached so the entire application shares one validated config.
-    """
-    return Settings()
-
-
-def setup_logging() -> None:
     """Configure root logging once at startup."""
     logging.config.dictConfig(LOGGING_CONFIG)
+    logger.debug("Logging configured")

src/transcription/db.py

@@ -4,93 +4,122 @@ V1 moves database resource ownership to explicit runtime initialization so
 startup/shutdown behavior is predictable and lifespan-managed.
 """
 
+from __future__ import annotations
+
 import contextlib
-from collections.abc import Generator
+import logging
+from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 
 from sqlalchemy import inspect
-from sqlalchemy.engine import Engine
+from sqlalchemy import text
-from sqlmodel import Session
+from sqlalchemy.engine import Connection
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.ext.asyncio import create_async_engine
 from sqlmodel import SQLModel
-from sqlmodel import create_engine
+from sqlmodel.ext.asyncio.session import AsyncSession
 
-from transcription.config import Settings
+from .config import Settings
-from transcription.config import get_settings
+from .config import get_settings
+
+logger = logging.getLogger(__name__)
 
 
 @dataclass(frozen=True)
 class DatabaseRuntime:
-    """Process-level database runtime resources."""
+    """Database runtime resources owned by app lifespan."""
 
-    engine: Engine
+    engine: AsyncEngine
+    session_factory: async_sessionmaker[AsyncSession]
 
 
 _runtime: DatabaseRuntime | None = None
 
 
-def _build_engine(settings: Settings) -> Engine:
+def _to_async_database_url(database_url: str) -> str:
+    """Normalize configured database URL to an async SQLAlchemy driver URL."""
+    if database_url.startswith("sqlite://") and not database_url.startswith("sqlite+aiosqlite://"):
+        return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1)
+    if database_url.startswith("postgresql://") and not database_url.startswith("postgresql+asyncpg://"):
+        return database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+    return database_url
+
+
+def _build_engine(settings: Settings) -> AsyncEngine:
+    database_url = _to_async_database_url(settings.database_url)
     connect_args: dict[str, object] = {}
-    if settings.database_url.startswith("sqlite"):
+    if database_url.startswith("sqlite"):
         connect_args["check_same_thread"] = False
-    return create_engine(
+    return create_async_engine(
-        settings.database_url,
+        url=database_url,
         echo=False,
+        pool_pre_ping=True,
         connect_args=connect_args,
     )
 
 
 def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime:
-    """Initialize and cache the process database runtime once."""
+    """Initialize lifespan-owned async DB resources once per process."""
     global _runtime
-
     if _runtime is not None:
         return _runtime
 
-    runtime_settings = settings or get_settings()
+    active_settings = settings or get_settings()
-    _runtime = DatabaseRuntime(engine=_build_engine(runtime_settings))
+    engine = _build_engine(active_settings)
+    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    _runtime = DatabaseRuntime(engine=engine, session_factory=session_factory)
+    logger.debug("Initialized async database runtime for database_url=%s", engine.url)
     return _runtime
 
 
-def get_database_runtime() -> DatabaseRuntime:
+def get_engine() -> AsyncEngine:
-    """Return initialized database runtime, creating it if needed."""
+    """Return the current async SQLAlchemy engine."""
-    if _runtime is None:
+    runtime = _runtime or initialize_database_runtime()
-        return initialize_database_runtime()
+    return runtime.engine
-    return _runtime
 
 
-def dispose_database_runtime() -> None:
+def get_session_factory() -> async_sessionmaker[AsyncSession]:
-    """Dispose process database runtime resources."""
+    """Return the shared async session factory."""
+    runtime = _runtime or initialize_database_runtime()
+    return runtime.session_factory
+
+
+async def cleanup_database() -> None:
+    """Cleanup database runtime resources."""
+    await dispose_database_runtime()
+
+
+async def dispose_database_runtime() -> None:
+    """Dispose lifespan-owned async database resources."""
     global _runtime
-    if _runtime is not None:
+    if _runtime is None:
-        _runtime.engine.dispose()
+        return
+    await _runtime.engine.dispose()
     _runtime = None
 
 
-def should_bootstrap_schema(settings: Settings) -> bool:
+async def create_all(*, engine: AsyncEngine | None = None) -> None:
-    """Return whether startup should auto-create schema for this environment."""
-    if settings.bootstrap_schema_on_startup is not None:
-        return settings.bootstrap_schema_on_startup
-    return settings.environment in {"development", "test"}
-
-
-def create_all(*, engine: Engine | None = None) -> None:
     """Create all tables on the selected engine."""
     # Import models so SQLModel metadata is fully registered before bootstrap.
     from transcription import models as _models  # noqa: F401
 
-    active_engine = engine or get_database_runtime().engine
+    active_engine = engine or get_engine()
-    SQLModel.metadata.create_all(active_engine)
+    async with active_engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+        await connection.run_sync(_ensure_sqlite_compat_columns)
+    logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url)
 
 
-def validate_schema_compatibility(*, engine: Engine | None = None) -> list[str]:
+def _ensure_sqlite_compat_columns(connection: Connection) -> None:
-    """Return schema compatibility issues for known V1 requirements.
+    """Apply lightweight dev/test SQLite compatibility column patches.
 
     This performs read-only validation and never mutates schema.
     """
-    active_engine = engine or get_database_runtime().engine
+    if connection.engine.url.get_backend_name() != "sqlite":
-    inspector = inspect(active_engine)
+        return
 
-    issues: list[str] = []
+    inspector = inspect(connection)
     table_names = set(inspector.get_table_names())
 
     required_tables = {"document", "job", "transcript", "transcriptrevision"}
@@ -98,17 +127,23 @@ def validate_schema_compatibility(*, engine: Engine | None = None) -> list[str]:
     for table_name in missing_tables:
         issues.append(f"missing_table:{table_name}")
 
-    if "job" in table_names:
+    columns = {column["name"] for column in inspector.get_columns("job")}
-        job_columns = {column["name"] for column in inspector.get_columns("job")}
+    if "retry_count" not in columns:
-        if "retry_count" not in job_columns:
+        connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
-            issues.append("missing_column:job.retry_count")
+        logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
-
-    return issues
 
 
-@contextlib.contextmanager
+@contextlib.asynccontextmanager
-def get_session(*, engine: Engine | None = None) -> Generator[Session]:
+async def get_session(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+) -> AsyncGenerator[AsyncSession]:
     """Yield a database session and ensure cleanup."""
-    active_engine = engine or get_database_runtime().engine
+    active_session_factory = session_factory or get_session_factory()
-    with Session(active_engine) as session:
+    async with active_session_factory() as session:
         yield session
+
+
+def should_bootstrap_schema(settings: Settings) -> bool:
+    """Compatibility helper for explicit bootstrap checks."""
+    return settings.should_bootstrap_schema

src/transcription/services/upload.py

@@ -8,7 +8,7 @@ from pathlib import Path
 from uuid import UUID
 from uuid import uuid4
 
-from sqlmodel import Session
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.config import Settings
 from transcription.config import get_settings
@@ -38,11 +38,11 @@ class UploadJobResult:
     original_filename: str
 
 
-def create_upload_job(
+async def create_upload_job(
     *,
     filename: str,
     file_bytes: bytes,
-    session: Session | None = None,
+    session: AsyncSession | None = None,
     settings: Settings | None = None,
 ) -> UploadJobResult:
     """Persist an uploaded file and create document/job records."""
@@ -70,10 +70,14 @@ def create_upload_job(
 
     try:
         if session is not None:
-            document, job = _create_upload_records(session=session, original_filename=filename, stored_path=stored_path)
+            document, job = await _create_upload_records(
+                session=session,
+                original_filename=filename,
+                stored_path=stored_path,
+            )
         else:
-            with get_session() as local_session:
+            async with get_session() as local_session:
-                document, job = _create_upload_records(
+                document, job = await _create_upload_records(
                     session=local_session,
                     original_filename=filename,
                     stored_path=stored_path,
@@ -133,22 +137,27 @@ def _build_stored_filename(filename: str) -> str:
     return f"{uuid4()}_{safe_name}"
 
 
-def _create_upload_records(*, session: Session, original_filename: str, stored_path: Path) -> tuple[Document, Job]:
+async def _create_upload_records(
+    *,
+    session: AsyncSession,
+    original_filename: str,
+    stored_path: Path,
+) -> tuple[Document, Job]:
     document = Document(
         filename=Path(original_filename).name,
         file_path=str(stored_path),
     )
     session.add(document)
-    session.flush()
+    await session.flush()
 
     job = Job(
         document_id=document.id,
         status=JobStatus.QUEUED,
     )
     session.add(job)
-    session.commit()
+    await session.commit()
-    session.refresh(document)
+    await session.refresh(document)
-    session.refresh(job)
+    await session.refresh(job)
     return document, job
 
 

src/transcription/ui/__init__.py

@@ -3,8 +3,8 @@
 from fastapi import FastAPI
 from nicegui import ui
 
-from transcription.ui.jobs_page import register_page as register_jobs_page
+from transcription.ui.pages.jobs_page import register_page as register_jobs_page
-from transcription.ui.upload_page import register_page as register_upload_page
+from transcription.ui.pages.upload_page import register_page as register_upload_page
 
 
 def register_pages(app: FastAPI) -> None:

src/transcription/ui/components/job_detail.py

@@ -0,0 +1,30 @@
+"""Reusable job detail rendering helpers."""
+
+from __future__ import annotations
+
+from nicegui import ui
+
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+
+
+def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None:
+    """Render all sections for the job detail page."""
+    ui.label(f"Job ID: {job.id}")
+    ui.label(f"Status: {job.status.value}")
+    ui.label(f"Created: {job.created_at.isoformat()}")
+    ui.label(f"Updated: {job.updated_at.isoformat()}")
+
+    if document is not None:
+        ui.label(f"Filename: {document.filename}")
+        ui.label(f"File path: {document.file_path}")
+
+    if transcript is None:
+        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

@@ -0,0 +1,55 @@
+"""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/pages/jobs_page.py

@@ -0,0 +1,92 @@
+"""Jobs list and detail page registration."""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+from nicegui import ui
+from sqlmodel import desc
+from sqlmodel import select
+
+from transcription.db import get_session
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+from transcription.ui.components.error_presenter import show_error
+from transcription.ui.components.error_presenter import summarize_error
+from transcription.ui.components.job_detail import render_job_detail
+from transcription.ui.components.job_table import JobTableRow
+from transcription.ui.components.job_table import render_jobs_table
+
+
+async def fetch_jobs() -> list[JobTableRow]:
+    """Return jobs for display in most-recent-first order."""
+    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:
+        ui.label("Transcription Jobs")
+        status = ui.label("Ready")
+
+        @ui.refreshable
+        async def render_table() -> None:
+            jobs = await fetch_jobs()
+            render_jobs_table(jobs)
+
+        async def refresh() -> None:
+            status.text = "Refreshing..."
+            try:
+                await render_table.refresh()
+                status.text = "Refreshed"
+            except Exception as exc:  # noqa: BLE001
+                status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}"
+                show_error(exc, title="Jobs refresh failed", operation="jobs.refresh")
+
+        ui.button("Refresh", on_click=refresh)
+        await render_table()
+        ui.link("Back to upload", "/")
+
+    @ui.page("/jobs/{job_id}")
+    async def job_detail_page(job_id: str) -> None:
+        ui.label("Job Detail")
+        try:
+            parsed_id = UUID(job_id)
+        except ValueError:
+            ui.label("Invalid job id")
+            ui.link("Back to jobs", "/jobs")
+            return
+
+        job, document, transcript = await fetch_job_detail(parsed_id)
+        if job is None:
+            ui.label("Job not found")
+            ui.link("Back to jobs", "/jobs")
+            return
+
+        render_job_detail(job=job, document=document, transcript=transcript)
+
+        ui.link("Back to jobs", "/jobs")

src/transcription/ui/pages/upload_page.py

@@ -10,8 +10,8 @@ from nicegui.events import UploadEventArguments
 from transcription.services.upload import UploadError
 from transcription.services.upload import UploadJobResult
 from transcription.services.upload import create_upload_job
-from transcription.ui.error_presenter import show_error
+from transcription.ui.components.error_presenter import show_error
-from transcription.ui.error_presenter import summarize_error
+from transcription.ui.components.error_presenter import summarize_error
 
 
 @dataclass
@@ -27,9 +27,9 @@ def accepted_upload_types() -> str:
     return ".jpg,.jpeg,.png,.tif,.tiff,.pdf"
 
 
-def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult:
+async def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult:
     """Create an upload job from incoming file data."""
-    return create_upload_job(filename=filename, file_bytes=file_bytes)
+    return await create_upload_job(filename=filename, file_bytes=file_bytes)
 
 
 def register_page() -> None:
@@ -49,7 +49,7 @@ def register_page() -> None:
             status_label.text = "Uploading..."
             try:
                 payload = await event.file.read()
-                result = submit_upload(filename=event.file.name, file_bytes=payload)
+                result = await submit_upload(filename=event.file.name, file_bytes=payload)
                 state.message = f"Created job {result.job_id}"
                 status_label.text = state.message
                 ui.notify(state.message, type="positive")

src/transcription/worker.py

@@ -2,16 +2,16 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
-import time
 from datetime import UTC
 from datetime import datetime
 from threading import Event
 
 from pydantic import ValidationError
-from sqlalchemy.engine import Engine
+from sqlalchemy.ext.asyncio import async_sessionmaker
-from sqlmodel import Session
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.config import Settings
 from transcription.config import get_settings
@@ -24,29 +24,28 @@ from transcription.models import Document
 from transcription.models import Job
 from transcription.models import JobStatus
 from transcription.models import Transcript
-from transcription.services.library import add_revision
 from transcription.services.transcription import transcribe_document_image
 
 logger = logging.getLogger(__name__)
 
 
-def process_next_queued_job(*, session: Session | None = None, engine: Engine | None = None) -> bool:
+async def process_next_queued_job(
+    *,
+    session: AsyncSession | None = None,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+) -> bool:
     """Process the next queued job and persist terminal outcome.
 
     Returns True when a job was processed, False when no queued job exists.
     """
     if session is None:
-        with get_session(engine=engine) as local_session:
+        async with get_session(session_factory=session_factory) as local_session:
-            return _process_next_queued_job(session=local_session)
+            return await _process_next_queued_job(session=local_session)
-    return _process_next_queued_job(session=session)
+    return await _process_next_queued_job(session=session)
 
 
-def _process_next_queued_job(*, session: Session) -> bool:
+async def _process_next_queued_job(*, session: AsyncSession) -> bool:
-    job = session.exec(
+    job = (await session.exec(select(Job).where(Job.status == JobStatus.QUEUED).order_by(Job.created_at))).first()
-        select(Job)
-        .where(Job.status == JobStatus.QUEUED)
-        .order_by(Job.created_at)
-    ).first()
 
     if job is None:
         return False
@@ -55,10 +54,10 @@ def _process_next_queued_job(*, session: Session) -> bool:
     job.status = JobStatus.PROCESSING
     job.updated_at = datetime.now(UTC)
     session.add(job)
-    session.commit()
+    await session.commit()
-    session.refresh(job)
+    await session.refresh(job)
 
-    document = session.get(Document, job.document_id)
+    document = await session.get(Document, job.document_id)
     if document is None:
         error = AppError(
             "Document not found",
@@ -76,17 +75,11 @@ def _process_next_queued_job(*, session: Session) -> bool:
 
     try:
         result = transcribe_document_image(document.file_path)
-        revision = add_revision(
+        await _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None)
-            job_id=job.id,
-            text=result.text,
-            source="worker",
-            accepted=False,
-            session=session,
-        )
         job.status = JobStatus.TRANSCRIBED
         job.updated_at = datetime.now(UTC)
         session.add(job)
-        session.commit()
+        await session.commit()
         logger.info(
             "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s revision_number=%s",
             job.id,
@@ -98,7 +91,7 @@ def _process_next_queued_job(*, session: Session) -> bool:
         error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job")
         settings = _get_worker_settings()
         if _should_retry(job=job, error=error, settings=settings):
-            _requeue_for_retry(session=session, job=job, error=error, settings=settings)
+            await _requeue_for_retry(session=session, job=job, error=error, settings=settings)
             logger.warning(
                 "Job retried operation=worker.process_job job_id=%s document_id=%s retry_count=%s error_id=%s category=%s",
                 job.id,
@@ -108,7 +101,7 @@ def _process_next_queued_job(*, session: Session) -> bool:
                 error.category.value,
             )
         else:
-            _finalize_failed_job(session=session, job=job, error=error)
+            await _finalize_failed_job(session=session, job=job, error=error)
             logger.exception(
                 "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
                 job.id,
@@ -120,16 +113,18 @@ def _process_next_queued_job(*, session: Session) -> bool:
     return True
 
 
-def _upsert_transcript(*, session: Session, job_id, text: str | None, error_detail: str | None) -> Transcript:
+async def _upsert_transcript(
-    transcript = session.exec(select(Transcript).where(Transcript.job_id == job_id)).first()
+    *, session: AsyncSession, job_id, text: str | None, error_detail: str | None
+) -> Transcript:
+    transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
     if transcript is None:
         transcript = Transcript(job_id=job_id)
 
     transcript.text = text
     transcript.error_detail = error_detail
     session.add(transcript)
-    session.commit()
+    await session.commit()
-    session.refresh(transcript)
+    await session.refresh(transcript)
     return transcript
 
 
@@ -144,32 +139,53 @@ def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool:
     return error.retriable and job.retry_count < settings.worker_max_retries
 
 
-def _requeue_for_retry(*, session: Session, job: Job, error: AppError, settings: Settings) -> None:
+async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
     job.retry_count += 1
     job.status = JobStatus.QUEUED
     job.updated_at = datetime.now(UTC)
     session.add(job)
-    session.commit()
+    await session.commit()
     if settings.worker_retry_backoff_seconds > 0:
-        time.sleep(settings.worker_retry_backoff_seconds)
+        await asyncio.sleep(settings.worker_retry_backoff_seconds)
 
 
-def _finalize_failed_job(*, session: Session, job: Job, error: AppError) -> None:
+async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
     job.status = JobStatus.FAILED
     job.updated_at = datetime.now(UTC)
     session.add(job)
-    session.commit()
+    await session.commit()
 
 
-def run_worker_loop(*, engine: Engine | None = None, stop_event: Event | None = None, poll_interval_seconds: float = 1.0) -> None:
+async def _run_worker_loop_async(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    stop_event: Event | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> None:
     """Run worker polling loop until stop_event is set."""
     while True:
         if stop_event is not None and stop_event.is_set():
             logger.info("Worker stop event received")
             return
 
-        processed = process_next_queued_job(engine=engine)
+        processed = await process_next_queued_job(session_factory=session_factory)
         if not processed:
-            time.sleep(poll_interval_seconds)
+            await asyncio.sleep(poll_interval_seconds)
+
+
+def run_worker_loop(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    stop_event: Event | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> None:
+    """Synchronous thread entrypoint that runs the async worker loop."""
+    asyncio.run(
+        _run_worker_loop_async(
+            session_factory=session_factory,
+            stop_event=stop_event,
+            poll_interval_seconds=poll_interval_seconds,
+        )
+    )