Merge branch 'main' into diverging

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 fastapi.responses import JSONResponse
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
 
-from transcription.api.errors import register_error_handlers
-from transcription.api.health import router as health_router
-from transcription.api.routes import router as transcription_router
-from transcription.config import get_settings
-from transcription.config import setup_logging
-from transcription.db import create_all
-from transcription.db import dispose_database_runtime
-from transcription.db import initialize_database_runtime
-from transcription.db import should_bootstrap_schema
-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
+from .api.errors import register_error_handlers
+from .api.health import router as health_router
+from .config import configure_logging
+from .config import get_settings
+from .db import cleanup_database
+from .db import create_all
+from .db import initialize_database_runtime
+from .ui import register_pages
+from .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):
-        create_all(engine=app.state.db_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}")
+    if settings.should_bootstrap_schema:
+        await create_all(engine=runtime.engine)
 
     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")
-    def _validate_operator_access_settings(self) -> "Settings":
-        if self.operator_access_enabled and not self.operator_password:
-            raise ValueError("OPERATOR_PASSWORD is required when OPERATOR_ACCESS_ENABLED=true")
-        return self
+    @property
+    def should_bootstrap_schema(self) -> bool:
+        """Return whether startup should auto-create schema for this environment."""
+        if self.bootstrap_schema_on_startup is not None:
+            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 get_settings() -> Settings:
-    """Return the singleton Settings instance.
-
-    Cached so the entire application shares one validated config.
-    """
-    return Settings()
-
-
-def setup_logging() -> None:
+def configure_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 sqlmodel import Session
+from sqlalchemy import text
+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 transcription.config import get_settings
+from .config import 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(
-        settings.database_url,
+    return create_async_engine(
+        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()
-    _runtime = DatabaseRuntime(engine=_build_engine(runtime_settings))
+    active_settings = settings or get_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:
-    """Return initialized database runtime, creating it if needed."""
+def get_engine() -> AsyncEngine:
+    """Return the current async SQLAlchemy engine."""
+    runtime = _runtime or initialize_database_runtime()
+    return runtime.engine
+
+
+def get_session_factory() -> async_sessionmaker[AsyncSession]:
+    """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 None:
-        return initialize_database_runtime()
-    return _runtime
-
-
-def dispose_database_runtime() -> None:
-    """Dispose process database runtime resources."""
-    global _runtime
-    if _runtime is not None:
-        _runtime.engine.dispose()
+        return
+    await _runtime.engine.dispose()
     _runtime = None
 
 
-def should_bootstrap_schema(settings: Settings) -> bool:
-    """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:
+async def create_all(*, engine: AsyncEngine | 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
-    SQLModel.metadata.create_all(active_engine)
+    active_engine = engine or get_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]:
-    """Return schema compatibility issues for known V1 requirements.
+def _ensure_sqlite_compat_columns(connection: Connection) -> None:
+    """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
-    inspector = inspect(active_engine)
+    if connection.engine.url.get_backend_name() != "sqlite":
+        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:
-        job_columns = {column["name"] for column in inspector.get_columns("job")}
-        if "retry_count" not in job_columns:
-            issues.append("missing_column:job.retry_count")
-
-    return issues
+    columns = {column["name"] for column in inspector.get_columns("job")}
+    if "retry_count" not in columns:
+        connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
+        logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
 
 
-@contextlib.contextmanager
-def get_session(*, engine: Engine | None = None) -> Generator[Session]:
+@contextlib.asynccontextmanager
+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
-    with Session(active_engine) as session:
+    active_session_factory = session_factory or get_session_factory()
+    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:
-                document, job = _create_upload_records(
+            async with get_session() as local_session:
+                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()
-    session.refresh(document)
-    session.refresh(job)
+    await session.commit()
+    await session.refresh(document)
+    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.upload_page import register_page as register_upload_page
+from transcription.ui.pages.jobs_page import register_page as register_jobs_page
+from transcription.ui.pages.upload_page import register_page as register_upload_page
 
 
 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.error_presenter import summarize_error
+from transcription.ui.components.error_presenter import show_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 sqlmodel import Session
+from sqlalchemy.ext.asyncio import async_sessionmaker
 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:
-            return _process_next_queued_job(session=local_session)
-    return _process_next_queued_job(session=session)
+        async with get_session(session_factory=session_factory) as local_session:
+            return await _process_next_queued_job(session=local_session)
+    return await _process_next_queued_job(session=session)
 
 
-def _process_next_queued_job(*, session: Session) -> bool:
-    job = session.exec(
-        select(Job)
-        .where(Job.status == JobStatus.QUEUED)
-        .order_by(Job.created_at)
-    ).first()
+async def _process_next_queued_job(*, session: AsyncSession) -> bool:
+    job = (await session.exec(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()
-    session.refresh(job)
+    await session.commit()
+    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(
-            job_id=job.id,
-            text=result.text,
-            source="worker",
-            accepted=False,
-            session=session,
-        )
+        await _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None)
         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:
-    transcript = session.exec(select(Transcript).where(Transcript.job_id == job_id)).first()
+async def _upsert_transcript(
+    *, 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()
-    session.refresh(transcript)
+    await session.commit()
+    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:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None:
+    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:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None:
+    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,
+        )
+    )