breaking up pages

src/transcription/app.py

@@ -7,24 +7,27 @@ from threading import Event
 from threading import Thread
 
 from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 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 .config import setup_logging
 from .db import cleanup_database
 from .db import create_all
-from .db import get_engine
+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_conn,
+            "session_factory": session_factory,
             "stop_event": stop_event,
             "poll_interval_seconds": 1.0,
         },
@@ -47,15 +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
-    engine = get_engine()
-    app.state.db_conn = engine
+    runtime = initialize_database_runtime(settings=settings)
+    app.state.db_engine = runtime.engine
+    app.state.db_session_factory = runtime.session_factory
 
     if settings.should_bootstrap_schema:
-        create_all(engine=engine)
+        await create_all(engine=runtime.engine)
 
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
@@ -65,7 +69,7 @@ async def _lifespan(app: FastAPI):
         yield
     finally:
         _stop_worker(app)
-        cleanup_database()
+        await cleanup_database()
 
 
 def create_app() -> FastAPI:

src/transcription/config.py

@@ -99,7 +99,7 @@ LOGGING_CONFIG: dict[str, object] = {
 }
 
 
-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,17 +4,21 @@ V1 moves database resource ownership to explicit runtime initialization so
 startup/shutdown behavior is predictable and lifespan-managed.
 """
 
+from __future__ import annotations
+
 import contextlib
 import logging
-from collections.abc import Generator
-from functools import cache
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
 
 from sqlalchemy import inspect
 from sqlalchemy import text
-from sqlalchemy.engine import Engine
-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 .config import Settings
 from .config import get_settings
@@ -22,66 +26,122 @@ from .config import get_settings
 logger = logging.getLogger(__name__)
 
 
-def _build_engine(settings: Settings) -> Engine:
+@dataclass(frozen=True)
+class DatabaseRuntime:
+    """Database runtime resources owned by app lifespan."""
+
+    engine: AsyncEngine
+    session_factory: async_sessionmaker[AsyncSession]
+
+
+_runtime: DatabaseRuntime | None = None
+
+
+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(
-        url=settings.database_url,
+    return create_async_engine(
+        url=database_url,
         echo=False,
+        pool_pre_ping=True,
         connect_args=connect_args,
     )
 
 
-@cache
-def get_engine() -> Engine:
-    """Return a new SQLAlchemy engine for the current settings."""
-    settings = get_settings()
-    logger.debug("Creating new SQLAlchemy engine for database_url=%s", settings.database_url)
-    return _build_engine(settings)
+def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime:
+    """Initialize lifespan-owned async DB resources once per process."""
+    global _runtime
+    if _runtime is not None:
+        return _runtime
+
+    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 cleanup_database() -> None:
+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."""
-    engine = get_engine()
-    engine.dispose()
+    await dispose_database_runtime()
 
 
-def create_all(*, engine: Engine | None = None) -> None:
+async def dispose_database_runtime() -> None:
+    """Dispose lifespan-owned async database resources."""
+    global _runtime
+    if _runtime is None:
+        return
+    await _runtime.engine.dispose()
+    _runtime = 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_engine()
-    SQLModel.metadata.create_all(active_engine)
-    _ensure_sqlite_compat_columns(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 _ensure_sqlite_compat_columns(engine: Engine) -> None:
+def _ensure_sqlite_compat_columns(connection: Connection) -> None:
     """Apply lightweight dev/test SQLite compatibility column patches.
 
     This keeps local bootstrap resilient when models evolve but no full
     migration tooling is in place yet.
     """
-    if engine.url.get_backend_name() != "sqlite":
+    if connection.engine.url.get_backend_name() != "sqlite":
         return
 
-    inspector = inspect(engine)
+    inspector = inspect(connection)
     table_names = set(inspector.get_table_names())
     if "job" not in table_names:
         return
 
     columns = {column["name"] for column in inspector.get_columns("job")}
     if "retry_count" not in columns:
-        with engine.begin() as connection:
-            connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
+        connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
         logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
 
 
-@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_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."""
@@ -66,10 +66,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,
@@ -122,22 +126,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/jobs_page.py

@@ -1,140 +0,0 @@
-"""Jobs list and detail page registration."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from uuid import UUID
-
-from nicegui import ui
-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.error_presenter import show_error
-from transcription.ui.error_presenter import summarize_error
-
-
-@dataclass(frozen=True)
-class JobView:
-    """Read model for rendering job rows in the UI."""
-
-    id: UUID
-    status: str
-    created_at: str
-    updated_at: str
-
-
-def fetch_jobs() -> list[JobView]:
-    """Return jobs for display in most-recent-first order."""
-    with get_session() as session:
-        jobs = session.exec(select(Job).order_by(Job.created_at.desc())).all()
-        return [
-            JobView(
-                id=job.id,
-                status=job.status.value,
-                created_at=job.created_at.isoformat(),
-                updated_at=job.updated_at.isoformat(),
-            )
-            for job in jobs
-        ]
-
-
-def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]:
-    """Return job, document, and transcript for detail view."""
-    with get_session() as session:
-        job = session.get(Job, job_id)
-        if job is None:
-            return None, None, None
-        document = session.get(Document, job.document_id)
-        transcript = 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")
-    def jobs_page() -> None:
-        ui.label("Transcription Jobs")
-        status = ui.label("Ready")
-        table_container = ui.column()
-
-        def render_table() -> None:
-            table_container.clear()
-            jobs = fetch_jobs()
-            with table_container:
-                if not jobs:
-                    ui.label("No jobs yet.")
-                    return
-                rows = [
-                    {
-                        "id": str(job.id),
-                        "status": job.status,
-                        "created_at": job.created_at,
-                        "updated_at": job.updated_at,
-                    }
-                    for job in jobs
-                ]
-                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=rows,
-                    row_key="id",
-                )
-                for row in rows:
-                    ui.link(f"Open {row['id']}", f"/jobs/{row['id']}")
-
-        def refresh() -> None:
-            status.text = "Refreshing..."
-            try:
-                render_table()
-                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)
-        render_table()
-        ui.link("Back to upload", "/")
-
-    @ui.page("/jobs/{job_id}")
-    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 = fetch_job_detail(parsed_id)
-        if job is None:
-            ui.label("Job not found")
-            ui.link("Back to jobs", "/jobs")
-            return
-
-        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)
-
-        ui.link("Back to jobs", "/jobs")

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
@@ -29,23 +29,23 @@ 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
@@ -54,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",
@@ -75,11 +75,11 @@ def _process_next_queued_job(*, session: Session) -> bool:
 
     try:
         result = transcribe_document_image(document.file_path)
-        _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None)
+        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",
             job.id,
@@ -90,7 +90,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,
@@ -100,7 +100,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,
@@ -112,16 +112,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
 
 
@@ -136,32 +138,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,
+        )
+    )