service updates

.dockerignore

@@ -0,0 +1,13 @@
+.git
+.gitignore
+.vscode
+.venv
+.pytest_cache
+.ruff_cache
+__pycache__/
+*.py[cod]
+*.db
+.env
+tests/
+docs/
+uploads/

Dockerfile

@@ -0,0 +1,47 @@
+FROM python:3.12-slim AS builder
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    UV_LINK_MODE=copy
+
+WORKDIR /app
+
+COPY --from=ghcr.io/astral-sh/uv:0.5.24 /uv /uvx /bin/
+
+COPY pyproject.toml uv.lock README.md ./
+RUN uv sync --frozen --no-dev --no-install-project
+
+COPY src ./src
+COPY prompts ./prompts
+RUN uv sync --frozen --no-dev
+
+
+FROM python:3.12-slim AS runtime
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PATH="/app/.venv/bin:$PATH" \
+    PYTHONPATH="/app/src" \
+    UPLOAD_DIR="/app/uploads" \
+    PROMPT_DIR="/app/prompts"
+
+WORKDIR /app
+
+RUN groupadd --system --gid 1001 appgroup \
+    && useradd --system --uid 1001 --gid appgroup --create-home appuser
+
+COPY --from=builder /app/.venv /app/.venv
+COPY --from=builder /app/src /app/src
+COPY --from=builder /app/prompts /app/prompts
+
+RUN mkdir -p /app/uploads /app/data \
+    && chown -R appuser:appgroup /app
+
+USER appuser
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=3s --start-period=3s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/healthz')"
+
+CMD ["uvicorn", "transcription.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000", "--proxy-headers"]

docker-compose.yml

@@ -0,0 +1,21 @@
+services:
+  transcription:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: transcription-app
+    env_file:
+      - .env
+    environment:
+      DATABASE_URL: sqlite:////app/data/transcription.db
+      UPLOAD_DIR: /app/uploads
+      PROMPT_DIR: /app/prompts
+    ports:
+      - "8002:8000"
+    volumes:
+      - ./uploads:/app/uploads
+      - transcription_data:/app/data
+    restart: unless-stopped
+
+volumes:
+  transcription_data:

docs/adr/ADR-0001-lifespan-owned-runtime-resources.md

@@ -1,35 +0,0 @@
-# ADR-0001: Lifespan-owned runtime resources
-
-- **Status:** accepted
-- **Date:** 2026-06-25
-
-## Context
-
-MVP initialized core runtime resources (database engine and worker dependencies) through module-level globals and startup side effects. `REQ-7` requires lifespan-owned runtime resources with explicit ownership and cleanup.
-
-## Decision
-
-Adopt lifespan-owned runtime resource initialization in `transcription.app`:
-
-1. Initialize database runtime during app lifespan startup.
-2. Store runtime handles on `app.state`.
-3. Pass runtime-owned dependencies (engine) to worker startup.
-4. Dispose runtime resources explicitly during lifespan shutdown.
-
-## Consequences
-
-### Positive
-- Explicit startup and shutdown ownership.
-- Predictable cleanup ordering.
-- Reduced hidden global side effects.
-
-### Tradeoffs
-- Minor wiring complexity in app startup.
-- Some call-sites still support fallback lazy initialization for compatibility.
-
-## Alternatives Considered
-
-1. **Keep module-level global ownership**
-   - Rejected: conflicts with `REQ-7` and increases ambiguity.
-2. **Introduce full async DB stack immediately**
-   - Rejected for Step 1: too broad for architecture-consolidation scope.

docs/adr/ADR-0002-explicit-schema-bootstrap-policy.md

@@ -1,36 +0,0 @@
-# ADR-0002: Explicit schema bootstrap policy
-
-- **Status:** accepted
-- **Date:** 2026-06-25
-
-## Context
-
-MVP called schema bootstrap (`create_all`) on every startup. `REQ-10` requires explicit, opt-in schema bootstrap behavior so normal production startup does not mutate schema.
-
-## Decision
-
-Add environment-aware bootstrap policy:
-
-1. New settings:
-   - `environment`: `development` | `test` | `production`
-   - `bootstrap_schema_on_startup`: optional explicit override
-2. Default behavior:
-   - Development/test: bootstrap enabled
-   - Production: bootstrap disabled
-3. App startup calls `create_all` only when policy evaluates true.
-
-## Consequences
-
-### Positive
-- Production startup behavior is safer and policy-driven.
-- Local development remains simple by default.
-
-### Tradeoffs
-- Deployments now require explicit schema management in production.
-
-## Alternatives Considered
-
-1. **Always bootstrap in all environments**
-   - Rejected: violates `REQ-10` intent.
-2. **Disable bootstrap everywhere immediately**
-   - Rejected: hurts local developer workflow without migration tool replacement yet.

docs/adr/ADR-0003-persistence-baseline-and-transition-path.md

@@ -1,31 +0,0 @@
-# ADR-0003: Persistence baseline and transition path
-
-- **Status:** accepted
-- **Date:** 2026-06-25
-
-## Context
-
-Architecture targets PostgreSQL baseline (optional MongoDB), while MVP currently runs on SQLite by default. V1 needs a clear transition path without destabilizing ongoing work.
-
-## Decision
-
-1. Preserve database URL configurability through centralized settings.
-2. Keep SQLite functional for local dev/test and fast feedback.
-3. Treat PostgreSQL as production baseline target for V1 completion.
-4. Keep persistence access behind `transcription.db` runtime/session access points.
-
-## Consequences
-
-### Positive
-- Clear migration path without immediate broad rewrite.
-- Controlled risk while preserving velocity.
-
-### Tradeoffs
-- Temporary dual-path assumptions (SQLite local vs PostgreSQL target).
-
-## Alternatives Considered
-
-1. **Immediate forced PostgreSQL-only migration**
-   - Rejected: higher short-term disruption risk.
-2. **Remain SQLite-only for V1**
-   - Rejected: inconsistent with architecture and requirement trajectory.

docs/adr/ADR-0004-in-process-worker-topology.md

@@ -1,32 +0,0 @@
-# ADR-0004: In-process worker topology for V1
-
-- **Status:** accepted
-- **Date:** 2026-06-25
-
-## Context
-
-The current system uses an in-process background worker. Architecture docs allow this in foundation stage and permit later hardening (optional external worker/queue).
-
-## Decision
-
-Retain in-process worker topology for V1, with improved lifecycle ownership:
-
-1. Worker starts/stops via app lifespan.
-2. Worker receives runtime-owned DB engine dependency explicitly.
-3. Extension path to external worker remains behind existing service/adapter seams.
-
-## Consequences
-
-### Positive
-- Keeps operational complexity low for personal-scale use.
-- Preserves delivery focus on V1 completion.
-
-### Tradeoffs
-- Throughput/scaling limits remain compared to external queue-based topology.
-
-## Alternatives Considered
-
-1. **Immediate queue/external worker introduction**
-   - Rejected: premature complexity for current scale.
-2. **Ad hoc thread lifecycle management outside lifespan**
-   - Rejected: weaker shutdown guarantees and poorer ownership clarity.

docs/adr/README.md

@@ -1,20 +0,0 @@
-# Architecture Decision Records (ADRs)
-
-This directory records significant architecture decisions for Version 1.
-
-## ADR Format
-
-Each ADR should include:
-
-1. **Status** (`proposed`, `accepted`, `superseded`)
-2. **Context**
-3. **Decision**
-4. **Consequences**
-5. **Alternatives Considered**
-
-## Index
-
-- [ADR-0001: Lifespan-owned runtime resources](ADR-0001-lifespan-owned-runtime-resources.md)
-- [ADR-0002: Explicit schema bootstrap policy](ADR-0002-explicit-schema-bootstrap-policy.md)
-- [ADR-0003: Persistence baseline and transition path](ADR-0003-persistence-baseline-and-transition-path.md)
-- [ADR-0004: In-process worker topology for V1](ADR-0004-in-process-worker-topology.md)

docs/ver1/ver1-step1-results.md

@@ -1,86 +0,0 @@
-# Ver1 Step 1 Results: Architecture Consolidation
-
-## Summary
-
-Step 1 implementation has been completed for the primary architecture-consolidation objectives:
-
-1. Lifespan-owned runtime resource model introduced for DB runtime ownership.
-2. Schema bootstrap policy changed from implicit-always to explicit/environment-aware.
-3. Worker startup now receives lifespan-owned DB engine dependency.
-4. ADR set established for key V1 architectural decisions.
-
-## Implemented Changes
-
-### 1) Runtime ownership
-
-- Updated `src/transcription/db.py`:
-  - Added `DatabaseRuntime` resource model.
-  - Added explicit runtime lifecycle methods:
-    - `initialize_database_runtime(...)`
-    - `get_database_runtime()`
-    - `dispose_database_runtime()`
-- Updated `src/transcription/app.py`:
-  - Lifespan initializes DB runtime and stores it on `app.state`.
-  - Lifespan disposes DB runtime on shutdown.
-
-### 2) Schema bootstrap policy (REQ-10 alignment)
-
-- Updated `src/transcription/config.py`:
-  - Added `environment` setting (`development`, `test`, `production`).
-  - Added `bootstrap_schema_on_startup` explicit override setting.
-- Updated `src/transcription/db.py`:
-  - Added `should_bootstrap_schema(settings)` policy function.
-- Updated `src/transcription/app.py`:
-  - Startup now calls `create_all(...)` only when policy allows.
-
-### 3) Worker dependency ownership
-
-- Updated `src/transcription/worker.py`:
-  - `process_next_queued_job(..., engine=None)` now supports explicit engine injection.
-  - `run_worker_loop(..., engine=None, ...)` now supports explicit engine injection.
-- Updated `src/transcription/app.py`:
-  - Worker thread is started with lifespan-owned engine.
-
-### 4) ADR governance
-
-Created:
-- `docs/adr/README.md`
-- `docs/adr/ADR-0001-lifespan-owned-runtime-resources.md`
-- `docs/adr/ADR-0002-explicit-schema-bootstrap-policy.md`
-- `docs/adr/ADR-0003-persistence-baseline-and-transition-path.md`
-- `docs/adr/ADR-0004-in-process-worker-topology.md`
-
-## Test Evidence
-
-Targeted regression checks executed successfully:
-
-- `uv run pytest tests/test_app.py tests/test_db.py tests/services/test_worker.py -q`
-- Result: pass
-
-## Residual Risks / Follow-ups
-
-1. Full REQ-7 completion may still require broader runtime ownership coverage for additional resources as V1 expands.
-2. Production schema management workflow (migrations/runbook tooling) should be finalized in subsequent V1 steps.
-3. Additional boundary enforcement automation (import-lint style checks) can be added in later hardening.
-
-## Step 1 Exit Assessment
-
-- Architecture ownership clarity: **met**
-- Schema bootstrap policy hardening: **met**
-- Worker lifecycle dependency clarity: **met**
-- ADR baseline established: **met**
-
-## Completion Checklist With Evidence
-
-| Criterion | Status | Evidence |
-| --- | --- | --- |
-| Architecture conformance matrix approved | partial | Consolidation implemented and documented in `docs/ver1/ver1-step1.md` + this results doc; formal matrix artifact can be added as a follow-up appendix. |
-| REQ-7 ownership gaps resolved or explicitly deferred | met | Lifespan-owned DB runtime and explicit worker engine wiring implemented in `src/transcription/app.py`, `src/transcription/db.py`, `src/transcription/worker.py`. Residual scope documented under follow-ups. |
-| REQ-10 explicit bootstrap policy implemented and verified | met | Policy implemented via `environment` + `bootstrap_schema_on_startup` in `src/transcription/config.py`, `should_bootstrap_schema(...)` in `src/transcription/db.py`, startup gate in `src/transcription/app.py`, tested in `tests/test_db.py`. |
-| Dependency direction rules documented and enforced | partial | Layering and runtime ownership documented in `docs/architecture.md`. Lightweight enforcement exists via review and test discipline; automated import-lint remains a follow-up. |
-| ADR set created for major Step 1 decisions | met | `docs/adr/README.md` and ADR-0001 through ADR-0004 created. |
-| Architecture/index docs updated to match implementation | met | `docs/architecture.md` and `docs/index.md` updated with V1 Step 1 runtime policy and links to V1/ADR artifacts. |
-| Regression and full test suites pass | met | Targeted: `uv run pytest tests/test_app.py tests/test_db.py tests/services/test_worker.py -q`; full suite: `uv run pytest -q`. |
-| Step 1 results artifact published | met | This document (`docs/ver1/ver1-step1-results.md`) created and updated with summary, evidence, risks, and checklist. |
-
-Step 1 is complete and ready to hand off to Ver1 Step 2.

docs/ver1/ver1-step1.md

@@ -1,309 +0,0 @@
-# Step 1 Implementation Plan: Architecture Consolidation
-
-## Purpose
-
-Align the implemented MVP codebase with the production architecture and V1 constraints documented in:
-
-- `docs/architecture.md`
-- `docs/requirements.md`
-- `docs/error_handling.md`
-- `docs/index.md`
-- `docs/intent.md`
-- `docs/ver1/ver1.md` (Step 1)
-
-This step hardens architecture boundaries and ownership without expanding product scope.
-
----
-
-## MCP Skill and Guide Inputs Incorporated
-
-This plan explicitly incorporates patterns and guardrails from john-stream-mcp resources:
-
-1. `resource://skills/fastapi-uv-docker/document`
-   - App factory and lifespan ownership
-   - Health endpoint and cloud-native baseline expectations
-   - Environment-driven configuration and startup discipline
-
-2. `resource://skills/fastapi-async-sqlalchemy-modernization/document`
-   - Current-state gap audit first
-   - Target runtime model before refactor
-   - Explicit resource lifecycle ownership
-   - Transaction/session boundary clarity
-   - Phased migration with rollback points
-
-3. `resource://skills/nicegui/document`
-   - Clear dependency direction
-   - UI/page registration as composition, not business logic container
-   - Async responsiveness and boundary separation
-
-4. `resource://prompts/greenfield-architecture/document`
-   - Pattern-comparison-first planning
-   - Explicit tradeoffs and staged implementation
-   - Output contract with risks, open questions, and next steps
-
----
-
-## Current-State Gap Summary (Architecture vs Implementation)
-
-Based on docs and current `src/transcription` code:
-
-1. **REQ-7 gap (lifespan-owned resources)**  
-   - DB engine/session factory are module globals in `db.py`, not app lifespan-owned.
-   - Worker thread lifecycle is owned by lifespan (good), but DB/provider resource ownership is mixed.
-
-2. **REQ-10 gap (explicit opt-in schema bootstrap)**  
-   - `create_all()` is executed unconditionally on startup in `app.py`.
-
-3. **Data store target gap (REQ-9 + architecture baseline)**  
-   - Runtime still defaults to SQLite MVP setup; production architecture targets PostgreSQL baseline with optional MongoDB.
-
-4. **Layering clarity gap (architecture layer model)**  
-   - Boundaries exist but are not yet formally enforced (interface/app/domain/infra dependency rules are implicit, not codified).
-
-5. **Decision record gap**  
-   - No ADR set documenting key V1 architectural decisions and deviations from MVP.
-
----
-
-## Scope for Step 1
-
-### In scope
-1. Produce architecture conformance audit and decision records.
-2. Define and implement target runtime ownership model for core resources.
-3. Establish explicit schema bootstrap policy (opt-in in production paths).
-4. Consolidate module boundaries and dependency direction rules.
-5. Update architecture docs to reflect implemented reality and V1 trajectory.
-
-### Out of scope
-- Full async SQLAlchemy rewrite (plan and seams only if deferred)
-- MongoDB feature implementation
-- New user-facing features
-- Major worker architecture replacement (in-process worker remains baseline)
-
----
-
-## Target Architecture Decisions for V1
-
-1. **Keep modular monolith topology** (FastAPI + NiceGUI + in-process worker).
-2. **Preserve container-light simplicity guardrails** from `architecture.md`.
-3. **Move runtime ownership to lifespan** for:
-   - DB engine/session factory lifecycle
-   - Worker runtime resources
-   - Provider client factory/config lifecycle
-4. **Adopt explicit schema bootstrap policy**:
-   - Dev/test: opt-in auto-bootstrap allowed
-   - Production: startup must not mutate schema implicitly
-5. **Formalize boundary map**:
-   - Interface (`api`, `ui`) -> Application (`services`) -> Domain (`models/rules`) -> Infrastructure (`db`, `providers`)
-   - No reverse imports
-
----
-
-## Detailed Work Breakdown
-
-## Phase A — Architecture Audit and Baseline Freeze
-
-- [ ] **A1. Produce architecture conformance matrix**
-  - Map each architecture section to current modules/files.
-  - Classify each row: `aligned`, `partial`, `not aligned`.
-
-- [ ] **A2. Produce REQ-7/REQ-9/REQ-10 focused gap report**
-  - Explicitly capture current vs required state.
-  - Include operational risk if left unresolved.
-
-- [ ] **A3. Freeze MVP architecture baseline**
-  - Record current baseline behavior and known temporary shortcuts.
-  - Link this baseline from `docs/ver1/ver1.md`.
-
-### Deliverables
-- `docs/ver1/ver1-step1-audit.md` (or equivalent section in this doc)
-- Architecture conformance table
-
-### Exit Criteria
-- No architecture changes begin before gap matrix and baseline are approved.
-
----
-
-## Phase B — Resource Ownership Consolidation (Lifespan-Centric)
-
-- [ ] **B1. Define runtime resource ownership contract**
-  - `app.py` lifespan owns resource initialization and cleanup order.
-  - `app.state` carries resource handles/factories.
-  - No hidden module-global side-effect initialization for runtime resources.
-
-- [ ] **B2. Refactor DB ownership model**
-  - Replace module-global engine singleton pattern with lifespan-initialized resource model.
-  - Define one canonical session-factory access path for app/worker/services.
-
-- [ ] **B3. Normalize worker dependencies**
-  - Ensure worker uses lifespan-owned resources/factories rather than implicit globals.
-  - Preserve deterministic startup/shutdown behavior.
-
-- [ ] **B4. Define provider adapter ownership**
-  - Provider client creation strategy is centralized and lifecycle-aware.
-  - Avoid per-call hidden client construction when unnecessary.
-
-### MCP-Guided Guardrails
-- Use explicit lifecycle composition patterns from `fastapi-async-sqlalchemy-modernization`.
-- Maintain app-factory + lifespan structure per `fastapi-uv-docker`.
-- Keep UI registration as composition only per `nicegui`.
-
-### Exit Criteria
-- Core runtime resources have one owner and one cleanup path.
-- No critical resource has ambiguous ownership.
-
----
-
-## Phase C — Schema Bootstrap Policy (REQ-10 Alignment)
-
-- [ ] **C1. Define environment-aware bootstrap policy**
-  - `auto_create_schema` (or equivalent) disabled in production by default.
-  - Startup schema mutation is explicit and intentional.
-
-- [ ] **C2. Split startup responsibilities**
-  - App startup performs health-critical initialization only.
-  - Schema bootstrap path is moved to explicit command/flag workflow.
-
-- [ ] **C3. Update deployment/runbook docs**
-  - Document migration/bootstrap flow for dev, staging, prod.
-  - Ensure policy is testable and auditable.
-
-### Exit Criteria
-- Normal production startup path does not call schema auto-create implicitly.
-- Bootstrap behavior is explicit and documented.
-
----
-
-## Phase D — Module Boundary Enforcement
-
-- [ ] **D1. Publish dependency direction rules**
-  - Allowed import directions across `api`, `ui`, `services`, `models/domain`, `db/providers`.
-  - Explicitly disallow reverse dependencies.
-
-- [ ] **D2. Reconcile package map with docs**
-  - Ensure docs’ architecture elements match real package layout and naming.
-  - Update docs where intentional deviations remain.
-
-- [ ] **D3. Isolate cross-layer responsibilities**
-  - Keep API/UI presentation concerns out of services.
-  - Keep provider/DB specifics out of interface layer.
-
-- [ ] **D4. Add lightweight architecture checks**
-  - Add static/import checks and/or review checklist in CI/review process.
-
-### Exit Criteria
-- Boundary rules are documented and applied.
-- Architectural drift can be detected during review/CI.
-
----
-
-## Phase E — Architecture Decision Records (ADRs)
-
-- [ ] **E1. Create ADR index**
-  - Add `docs/adr/README.md` with template and status model.
-
-- [ ] **E2. Record minimum V1 ADR set**
-  1. Runtime ownership model (lifespan-owned resources)
-  2. Schema bootstrap policy (explicit vs implicit)
-  3. Persistence baseline (PostgreSQL target; SQLite transition strategy)
-  4. Worker topology (in-process for V1, extension path preserved)
-
-- [ ] **E3. Cross-link ADRs**
-  - Link from architecture and V1 docs.
-
-### Exit Criteria
-- Major architecture decisions are explicit, versioned, and discoverable.
-
----
-
-## Phase F — Documentation Consolidation
-
-- [ ] **F1. Update `docs/architecture.md`**
-  - Reflect real implementation and V1 target state separately.
-  - Mark transitional choices clearly.
-
-- [ ] **F2. Update `docs/index.md` navigation consistency**
-  - Ensure architecture/readme references match actual docs/files.
-
-- [ ] **F3. Update `docs/requirements.md` traceability notes**
-  - Mark REQ-7/REQ-10 status and verification approach after consolidation.
-
-- [ ] **F4. Add Step 1 result summary**
-  - Create `docs/ver1/ver1-step1-results.md` after implementation.
-
-### Exit Criteria
-- Docs are internally consistent and match runtime architecture reality.
-
----
-
-## Verification Plan
-
-## Architecture Verification Matrix (Step 1)
-
-1. **Inspection**
-   - Resource ownership map exists and matches code.
-   - Schema bootstrap policy is explicit and environment-aware.
-   - ADRs exist for each key architecture decision.
-
-2. **Automated checks**
-   - Existing test suite remains green.
-   - New/updated tests validate startup policy (no implicit schema mutation in production mode).
-   - Import/dependency-direction checks pass (if introduced).
-
-3. **Demonstration**
-   - App starts in dev mode with explicit expected behavior.
-   - App starts in production mode without mutating schema implicitly.
-   - Worker lifecycle starts/stops cleanly with app lifespan.
-
----
-
-## Risks and Mitigations
-
-1. **Risk:** Refactor destabilizes MVP behavior  
-   **Mitigation:** Phase changes with small PRs and regression checks after each phase.
-
-2. **Risk:** Over-rotation into premature async rewrite  
-   **Mitigation:** Keep this step focused on lifecycle ownership and boundaries; defer full async migration unless required.
-
-3. **Risk:** Schema policy changes break local DX  
-   **Mitigation:** Keep explicit dev bootstrap path simple and documented.
-
-4. **Risk:** Boundary rules become “doc only”  
-   **Mitigation:** Add CI/review enforcement and architecture checklist.
-
----
-
-## Recommended Implementation Order
-
-1. Phase A — Audit and baseline freeze  
-2. Phase B — Resource ownership consolidation  
-3. Phase C — Schema bootstrap policy  
-4. Phase D — Boundary enforcement  
-5. Phase E — ADR authoring  
-6. Phase F — Documentation consolidation
-
-This order minimizes risk: diagnose first, then refactor ownership, then lock policy, then enforce boundaries, and finally finalize docs.
-
----
-
-## Step 1 Completion Checklist
-
-- [ ] Architecture conformance matrix approved.
-- [ ] REQ-7 ownership gaps resolved or explicitly deferred with owner/date.
-- [ ] REQ-10 explicit bootstrap policy implemented and verified.
-- [ ] Dependency direction rules documented and enforced.
-- [ ] ADR set created for all major Step 1 decisions.
-- [ ] Architecture and index docs updated to match implementation.
-- [ ] Full test suite passes after consolidation.
-- [ ] `docs/ver1/ver1-step1-results.md` created with evidence and residual risks.
-
----
-
-## Handoff to Step 2
-
-Once Step 1 completes, Step 2 (Error Handling & Reliability Hardening) can proceed on stable architecture seams:
-
-- consistent lifecycle ownership,
-- explicit startup policy,
-- clear module boundaries,
-- documented architecture decisions.

docs/ver1/ver1-step2-results.md

@@ -1,80 +0,0 @@
-# Ver1 Step 2 Results: Error Handling & Reliability Hardening
-
-## Summary
-
-Step 2 implementation is complete for the planned reliability and error-handling hardening scope:
-
-1. Worker retries are now explicit, bounded, and category-driven.
-2. Error behavior is more consistent across worker/API/UI boundaries.
-3. Logging now includes stronger boundary context in key failure paths.
-4. Test coverage was expanded for retry policy and new reliability settings.
-
-## Implemented Changes
-
-### 1) Worker retry policy and terminal behavior
-
-- Updated `src/transcription/models.py`:
-  - Added `Job.retry_count` with default `0`.
-- Updated `src/transcription/config.py`:
-  - Added `worker_max_retries`.
-  - Added `worker_retry_backoff_seconds`.
-- Updated `src/transcription/worker.py`:
-  - Added bounded retry decision path (`_should_retry`).
-  - Added requeue behavior (`_requeue_for_retry`) for retriable errors.
-  - Added deterministic terminal failure behavior (`_finalize_failed_job`).
-  - Preserved transcript failure detail persistence (`error_id`, `category`, suggestion).
-
-### 2) API fallback normalization hardening
-
-- Updated `src/transcription/api/errors.py`:
-  - Fallback handler now emits safe generic internal message for unhandled exceptions.
-  - Added structured boundary logging fields including operation and exception type.
-
-### 3) UI interaction reliability guard
-
-- Updated `src/transcription/ui/upload_page.py`:
-  - Added duplicate in-flight submission guard to prevent repeated upload handling while busy.
-
-### 4) Observability/logging improvements
-
-- Updated worker logs in `src/transcription/worker.py` to include operation and domain identifiers in key transitions:
-  - pick
-  - retry
-  - transcribed
-  - failed
-
-## Test Coverage Added/Updated
-
-- Updated `tests/test_models.py`:
-  - Assert `retry_count` default.
-- Updated `tests/test_config.py`:
-  - Added worker retry settings default test.
-- Updated `tests/services/test_worker.py`:
-  - Added retriable requeue test.
-  - Added retry-exhaustion terminal failure test.
-  - Updated existing tests for settings-driven worker behavior.
-- Existing API error tests remained green with fallback behavior updates:
-  - `tests/api/test_error_responses.py`
-
-## Verification Evidence
-
-Executed and passing:
-
-- `uv run pytest tests/services/test_worker.py tests/test_models.py tests/test_config.py tests/api/test_error_responses.py -q`
-- `uv run pytest -q`
-
-## Residual Risks / Follow-ups
-
-1. Retry policy currently uses simple fixed backoff; richer strategy (exponential/jitter) can be added in later hardening.
-2. Full cross-layer structured logging standardization can be expanded in Step 6 observability work.
-3. A formal Step 2 error-path inventory artifact (`ver1-step2-audit.md`) is still recommended for governance completeness.
-
-## Step 2 Exit Assessment
-
-- Error taxonomy and envelope stability: **met**
-- Bounded retry and terminal failure behavior: **met**
-- Worker reliability controls: **met**
-- UI interaction hardening for duplicate actions: **met**
-- Test coverage expansion and full-suite regression safety: **met**
-
-Step 2 is complete and ready to hand off to Ver1 Step 3.

docs/ver1/ver1-step2.md

@@ -1,302 +0,0 @@
-# Step 2 Implementation Plan: Error Handling & Reliability Hardening
-
-## Purpose
-
-Implement **Ver1 Step 2** from `docs/ver1/ver1.md` by standardizing failure behavior and reliability controls so the system fails safely, predictably, and transparently across UI, API, services, worker, and provider boundaries.
-
-Primary governing docs:
-
-- `docs/error_handling.md` (authoritative contract)
-- `docs/requirements.md` (REQ-2, REQ-3, REQ-4, REQ-5, REQ-6)
-- `docs/architecture.md` (boundary ownership and worker lifecycle)
-- `docs/ver1/ver1.md` (Step 2 objective)
-
----
-
-## MCP Skill and Guide Inputs Incorporated
-
-This plan integrates guidance from john-stream-mcp resources:
-
-1. `resource://skills/python-logging-dictconfig/document`
-   - centralized `dictConfig` logging
-   - startup-only configuration
-   - stable named loggers and boundary-level logging discipline
-
-2. `resource://skills/pytesting/document`
-   - deterministic, behavior-first tests
-   - explicit marker usage and fast/slow lane discipline
-   - integration checks for boundary behavior and error contracts
-
-3. `resource://skills/fastapi-async-sqlalchemy-modernization/document`
-   - classify at source boundary
-   - explicit transaction/session behavior under failure
-   - phased rollout with quality gates and rollback awareness
-
-4. `resource://skills/nicegui-ui-customization/document`
-   - explicit user-facing error feedback for each interaction
-   - prevent duplicate actions during in-flight operations
-   - preserve one-way dependency boundaries from UI -> services
-
-5. `resource://skills/fastapi-uv-docker/document` (applied selectively)
-   - lifespan-safe startup/shutdown behavior
-   - health/readiness posture and cloud-native operational checks
-
----
-
-## Current-State Gap Summary
-
-The project already has a strong baseline (`AppError`, taxonomy enum, API envelope, worker persistence), but Step 2 needs completion-level hardening:
-
-1. **Error contract consistency**
-   - API envelope exists, but consistency must be verified for all error pathways.
-2. **Cross-boundary category normalization**
-   - Provider/service/worker mappings exist, but require stricter policy checks and tests.
-3. **Retry policy implementation depth**
-   - Step 2 requires bounded retry policy and clear terminal behavior for retriable failures.
-4. **Operational traceability**
-   - Logging exists; Step 2 requires consistent structured fields at critical boundaries.
-5. **UI failure UX consistency**
-   - UI error handling exists; Step 2 requires explicit contract coverage and anti-duplication safeguards.
-
----
-
-## Scope for Step 2
-
-### In scope
-1. Enforce canonical error taxonomy and envelope across all boundaries.
-2. Standardize logging fields and boundary-level error traceability.
-3. Implement/complete bounded retry and terminal failure behavior in worker paths.
-4. Improve UI/API error presentation consistency and actionable guidance.
-5. Add comprehensive Step 2 test coverage and verification matrix.
-6. Update documentation to reflect final Step 2 policies and behavior.
-
-### Out of scope
-- Major architecture/topology changes (external queue, distributed worker)
-- New end-user feature expansion outside reliability/error handling
-- Full async ORM migration (unless required by bug fix)
-
----
-
-## Target Decisions for Step 2
-
-1. **Taxonomy stability is mandatory**
-   - `ErrorCategory` values remain stable contract identifiers.
-2. **Classification occurs at source boundary**
-   - adapters/services normalize early; UI/API only present safely.
-3. **User safety over internal detail leakage**
-   - expose safe message + suggestion + error_id; keep sensitive detail in logs.
-4. **Retry is explicit and bounded**
-   - only retriable categories may retry; retries are capped; terminal failures persist reason.
-5. **Boundary logs carry correlation fields**
-   - include `error_id`, `category`, `operation`, and domain identifiers where available.
-
----
-
-## Detailed Work Breakdown
-
-## Phase A — Error Contract Audit and Policy Lock
-
-- [ ] **A1. Build error-path inventory**
-  - Enumerate all failure entry points across:
-    - `api/`
-    - `ui/`
-    - `services/`
-    - `worker.py`
-    - `providers/`
-
-- [ ] **A2. Produce taxonomy mapping table**
-  - For each known exception path, map:
-    - source exception type
-    - target `ErrorCategory`
-    - retriable flag
-    - API status (if exposed)
-
-- [ ] **A3. Reconcile with `docs/error_handling.md`**
-  - Resolve any mismatch in category semantics, status codes, or suggested actions.
-
-### Deliverables
-- `docs/ver1/ver1-step2-audit.md` (recommended)
-- taxonomy mapping table
-
-### Exit Criteria
-- Every known failure path has explicit category + retriable policy.
-
----
-
-## Phase B — API and Service Contract Hardening
-
-- [ ] **B1. Enforce API envelope completeness**
-  - Ensure all API errors return:
-    - `error_id`, `category`, `message`, `suggestion`, `timestamp`
-
-- [ ] **B2. Verify category-to-status mapping consistency**
-  - Confirm `api/errors.py` matches `docs/error_handling.md` mapping guidance.
-
-- [ ] **B3. Normalize service exceptions at boundary**
-  - Services should raise `AppError` subclasses for known failures.
-  - Unknown exceptions must become `internal_unexpected_error` with traceable `error_id`.
-
-- [ ] **B4. Ensure safe detail handling**
-  - API/UI messages remain safe.
-  - Diagnostic context remains in logs/persisted failure detail where appropriate.
-
-### Exit Criteria
-- No unstructured/unclassified exception escapes core boundaries.
-- API responses are contract-stable for all tested failure modes.
-
----
-
-## Phase C — Worker Retry and Terminal Failure Policy
-
-- [ ] **C1. Define bounded retry policy**
-  - Add configurable retry settings (attempt limit/backoff policy).
-  - Limit retries to retriable categories.
-
-- [ ] **C2. Implement terminal failure persistence**
-  - On retry exhaustion, persist clear terminal reason and `error_id`.
-  - Ensure job status transitions end deterministically at `failed`.
-
-- [ ] **C3. Add duplicate-processing safety checks**
-  - Prevent duplicate terminal updates when job already resolved.
-
-- [ ] **C4. Validate worker lifecycle under repeated transient failures**
-  - Ensure loop remains stable and responsive.
-
-### Exit Criteria
-- Retries are bounded and policy-driven.
-- Exhausted retries produce deterministic failed state with evidence.
-
----
-
-## Phase D — Logging and Observability Contract Enforcement
-
-- [ ] **D1. Central logging conformance check**
-  - Confirm startup-only `dictConfig` use remains canonical.
-  - No module-level `basicConfig` use.
-
-- [ ] **D2. Standardize error log fields**
-  - Require at minimum when available:
-    - `error_id`, `category`, `operation`, `exception_type`, `job_id`, `document_id`
-
-- [ ] **D3. Boundary handoff logging**
-  - Add/normalize logs at transitions:
-    - UI action -> service
-    - service -> provider/db
-    - worker pickup -> terminal state
-
-- [ ] **D4. Log noise control**
-  - Avoid duplicate stack-trace logging across layers for same exception.
-
-### Exit Criteria
-- Critical failure events are traceable end-to-end via logs and `error_id`.
-
----
-
-## Phase E — UI Error UX Consistency and Interaction Hardening
-
-- [ ] **E1. Standardize user error presentation**
-  - For upload/jobs interactions, ensure:
-    - clear title
-    - plain-language message
-    - suggested action
-    - visible error reference id
-
-- [ ] **E2. Add in-flight interaction guards**
-  - Prevent duplicate submits/click storms during pending operations.
-
-- [ ] **E3. Ensure deterministic UI state recovery**
-  - controls re-enable after failure
-  - status text remains actionable
-
-- [ ] **E4. Keep UI boundary clean**
-  - no provider/protocol details leaked into page modules
-
-### Exit Criteria
-- All primary UI actions have consistent success/failure interaction behavior.
-
----
-
-## Phase F — Test Expansion and Verification
-
-Apply pytesting guidance: behavior-first assertions, deterministic fixtures, strict markers.
-
-- [ ] **F1. API error contract tests**
-  - verify envelope fields and status mapping for each category class.
-
-- [ ] **F2. Service classification tests**
-  - verify known failures map to expected `AppError` subclasses/categories.
-
-- [ ] **F3. Worker retry policy tests**
-  - retriable failure retries and eventual success
-  - retriable failure exhaustion -> terminal failed
-  - non-retriable failure -> immediate failed
-
-- [ ] **F4. UI error behavior tests**
-  - upload/jobs actions show actionable feedback on failures
-  - duplicate action guard behavior
-
-- [ ] **F5. Regression guard tests**
-  - at least one test per previously observed production/real-world failure mode
-
-### Validation Commands
-- `uv run pytest --collect-only -q`
-- `uv run pytest -m unit -q`
-- `uv run pytest -m "not external" -q`
-- `uv run pytest -q`
-
-### Exit Criteria
-- All Step 2 reliability/error contract tests pass.
-- Existing suite remains green.
-
----
-
-## Recommended Implementation Order
-
-1. Phase A — audit and policy lock  
-2. Phase B — API/service contract hardening  
-3. Phase C — worker retry and terminal policy  
-4. Phase D — logging/traceability normalization  
-5. Phase E — UI consistency hardening  
-6. Phase F — test expansion and full verification
-
-This order reduces risk by locking policy first, then applying behavior changes at core boundaries before UI polish.
-
----
-
-## Risks and Mitigations
-
-1. **Risk:** Overly broad retry policy causes hidden failure loops  
-   **Mitigation:** strict category-based retry eligibility + hard cap + terminal persistence.
-
-2. **Risk:** User-facing messages become too technical  
-   **Mitigation:** enforce safe message + suggestion contract in tests.
-
-3. **Risk:** Logging becomes noisy/redundant  
-   **Mitigation:** boundary logging rules and single-trace ownership.
-
-4. **Risk:** Reliability work introduces regressions in happy path  
-   **Mitigation:** run full suite continuously; preserve integration pipeline tests.
-
----
-
-## Step 2 Completion Checklist
-
-- [ ] Error taxonomy mapping table completed and approved.
-- [ ] API envelope and HTTP status behavior verified for all relevant failure categories.
-- [ ] Service/provider exception normalization is consistent and tested.
-- [ ] Worker retry behavior is bounded, explicit, and terminal-state safe.
-- [ ] Structured error logging fields are present at boundary handoffs.
-- [ ] UI failure flows provide clear, actionable, and traceable feedback.
-- [ ] Full test suite passes with new Step 2 coverage included.
-- [ ] `docs/ver1/ver1-step2-results.md` created with evidence and residual risks.
-
----
-
-## Handoff to Step 3
-
-After Step 2 completion, Step 3 (Functional Completion by Requirement Domain) proceeds on a hardened foundation:
-
-- stable failure contracts,
-- predictable retries and terminal behavior,
-- actionable user/API error semantics,
-- improved diagnostic traceability.

pyproject.toml

@@ -12,9 +12,12 @@ description = "Historical document transcription system"
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
+    "aiosqlite>=0.21.0",
+    "asyncpg>=0.31.0",
     "fastapi>=0.138.0",
     "nicegui==3.13.0",
     "openrouter>=0.7.0",
+    "psycopg2-binary>=2.9.12",
     "pydantic>=2.13.4",
     "pydantic-settings>=2.9.1",
     "sqlmodel>=0.0.25",
@@ -26,6 +29,13 @@ dev = [
     "pytest-asyncio>=0.25",
 ]
 
+[dependency-groups]
+dev = [
+    "pre-commit>=4.6.0",
+    "ruff>=0.15.20",
+    "ty>=0.0.54",
+]
+
 [tool.pytest.ini_options]
 addopts = "--strict-markers -q"
 markers = [

ruff.toml

@@ -0,0 +1,62 @@
+line-length = 120
+indent-width = 4
+target-version = "py313"
+
+exclude = [
+    ".venv",
+    ".devenv",
+    ".git",
+    ".vscode",
+    "build",
+    "site",
+    "__pycache__",
+]
+
+[lint]
+preview = true
+
+extend-select = [
+    "ARG",      # https://docs.astral.sh/ruff/rules/#flake8-unused-arguments-arg
+	"B",        # https://docs.astral.sh/ruff/rules/#flake8-bugbear-b
+	"C4",       # https://docs.astral.sh/ruff/rules/#flake8-comprehensions-c4
+    "DOC102",   # https://docs.astral.sh/ruff/rules/docstring-extraneous-parameter/
+    "DOC202",   # https://docs.astral.sh/ruff/rules/docstring-extraneous-returns/
+    "DOC403",   # https://docs.astral.sh/ruff/rules/docstring-extraneous-yields/
+    "DOC502",   # https://docs.astral.sh/ruff/rules/docstring-extraneous-exception/
+	"E", "W",   # https://docs.astral.sh/ruff/rules/#pycodestyle-e-w
+	"F",        # https://docs.astral.sh/ruff/rules/#pyflakes-f
+    "FURB",     # https://docs.astral.sh/ruff/rules/#refurb-furb
+	"I",        # https://docs.astral.sh/ruff/rules/#isort-i
+    "N",        # https://docs.astral.sh/ruff/rules/#pep8-naming-n
+    "PD",       # https://docs.astral.sh/ruff/rules/#pandas-vet-pd
+    "PTH",      # https://docs.astral.sh/ruff/rules/#flake8-use-pathlib-pth
+	"UP",       # https://docs.astral.sh/ruff/rules/#pyupgrade-up
+	"SIM",      # https://docs.astral.sh/ruff/rules/#flake8-simplify-sim
+    "PLR0202",  # https://docs.astral.sh/ruff/rules/no-classmethod-decorator/
+    "PLR0203",  # https://docs.astral.sh/ruff/rules/no-staticmethod-decorator/
+    "PLR0206",  # https://docs.astral.sh/ruff/rules/property-with-parameters/
+    "PLR0915",  # https://docs.astral.sh/ruff/rules/too-many-statements/
+    "PLR1702",  # https://docs.astral.sh/ruff/rules/too-many-nested-blocks/
+    "TRY002",
+]
+extend-fixable = ["ALL"]
+ignore = [
+    "UP046",
+    "UP047",
+]
+
+[lint.extend-per-file-ignores]
+"*.ipynb" = [
+    "F401", # unused imports
+    "F841", # unused local variable
+    "F821", # undefined name in exploratory notebook cells
+]
+
+[lint.isort]
+force-single-line = true
+
+[format]
+quote-style = "double"
+indent-style = "space"
+skip-magic-trailing-comma = false
+line-ending = "auto"

src/transcription/app.py

@@ -2,50 +2,57 @@
 
 from __future__ import annotations
 
+import asyncio
 from contextlib import asynccontextmanager
-from threading import Event
+from contextlib import suppress
-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 .db import cleanup_database
 from .db import create_all
+from .db import dispose_database_runtime
 from .db import initialize_database_runtime
+from .services import ServiceBundle
 from .ui import register_pages
 from .worker import run_worker_loop
 
 
 def _start_worker(app: FastAPI) -> None:
-    session_factory: async_sessionmaker[AsyncSession] = app.state.db_session_factory
+    stop_event = asyncio.Event()
-    stop_event = Event()
+    wake_queue: asyncio.Queue[None] = asyncio.Queue()
-    worker_thread = Thread(
+    worker_task = asyncio.create_task(
-        target=run_worker_loop,
+        run_worker_loop(
-        kwargs={
+            session_factory=app.state.runtime.session_factory,
-            "session_factory": session_factory,
+            stop_event=stop_event,
-            "stop_event": stop_event,
+            wake_queue=wake_queue,
-            "poll_interval_seconds": 1.0,
+            poll_interval_seconds=1.0,
-        },
+        )
-        daemon=True,
     )
-    worker_thread.start()
+    wake_queue.put_nowait(None)
     app.state.worker_stop_event = stop_event
-    app.state.worker_thread = worker_thread
+    app.state.worker_wake_queue = wake_queue
+    app.state.worker_task = worker_task
 
 
-def _stop_worker(app: FastAPI) -> None:
+async def _stop_worker(app: FastAPI) -> None:
     stop_event = getattr(app.state, "worker_stop_event", None)
-    worker_thread = getattr(app.state, "worker_thread", None)
+    wake_queue = getattr(app.state, "worker_wake_queue", None)
+    worker_task = getattr(app.state, "worker_task", None)
 
     if stop_event is not None:
         stop_event.set()
-    if worker_thread is not None:
+    if wake_queue is not None:
-        worker_thread.join(timeout=2.0)
+        wake_queue.put_nowait(None)
+    if worker_task is not None:
+        try:
+            await asyncio.wait_for(worker_task, timeout=2.0)
+        except TimeoutError:
+            worker_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await worker_task
 
 
 @asynccontextmanager
@@ -54,12 +61,11 @@ async def _lifespan(app: FastAPI):
 
     settings = get_settings()
     app.state.settings = settings
-    runtime = initialize_database_runtime(settings=settings)
+    app.state.services = ServiceBundle()
-    app.state.db_engine = runtime.engine
+    app.state.runtime = initialize_database_runtime(settings=settings)
-    app.state.db_session_factory = runtime.session_factory
 
     if settings.should_bootstrap_schema:
-        await create_all(engine=runtime.engine)
+        await create_all(engine=app.state.runtime.engine)
 
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
@@ -68,8 +74,8 @@ async def _lifespan(app: FastAPI):
     try:
         yield
     finally:
-        _stop_worker(app)
+        await _stop_worker(app)
-        await cleanup_database()
+        await dispose_database_runtime()
 
 
 def create_app() -> FastAPI:

src/transcription/config.py

@@ -41,6 +41,7 @@ class Settings(BaseSettings):
     # --- persistence ---
     database_url: str = "sqlite:///./transcription.db"
     bootstrap_schema_on_startup: bool | None = None
+    sqlite_check_same_thread: bool = False
 
     # --- filesystem paths ---
     upload_dir: Path = Path("./uploads")
@@ -61,10 +62,10 @@ class Settings(BaseSettings):
 _settings: ContextVar[Settings | None] = ContextVar("settings", default=None)
 
 
-def get_settings() -> Settings:
+def get_settings(**kwargs) -> Settings:
     settings = _settings.get()
     if settings is None:
-        settings = Settings()  # pyright: ignore[reportCallIssue]
+        settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
         _settings.set(settings)
     return settings
 

src/transcription/db.py

@@ -1,147 +0,0 @@
-"""Database runtime ownership, schema bootstrap, and session access.
-
-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 AsyncGenerator
-from dataclasses import dataclass
-
-from sqlalchemy import inspect
-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.ext.asyncio.session import AsyncSession
-
-from .config import Settings
-from .config import get_settings
-
-logger = logging.getLogger(__name__)
-
-
-@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 database_url.startswith("sqlite"):
-        connect_args["check_same_thread"] = False
-    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 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 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
-    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()
-    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(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 connection.engine.url.get_backend_name() != "sqlite":
-        return
-
-    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:
-        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.asynccontextmanager
-async def get_session(
-    *,
-    session_factory: async_sessionmaker[AsyncSession] | None = None,
-) -> AsyncGenerator[AsyncSession]:
-    """Yield a database session and ensure cleanup."""
-    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/db/__init__.py

@@ -0,0 +1,6 @@
+from .operations import create_all
+from .runtime import dispose_database_runtime
+from .runtime import get_session
+from .runtime import initialize_database_runtime
+
+__all__ = ["create_all", "dispose_database_runtime", "get_session", "initialize_database_runtime"]

src/transcription/db/operations.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import inspect
+from sqlalchemy import text
+from sqlalchemy.engine import Connection
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlmodel import SQLModel
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..models import Job
+from ..models import JobStatus
+from .runtime import get_engine
+
+logger = logging.getLogger(__name__)
+
+
+async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
+    """Get the next queued job, if any."""
+    result = await session.exec(
+        select(Job)
+        .where(Job.status == JobStatus.QUEUED)
+        .order_by(Job.created_at)  # pyright: ignore[reportArgumentType]
+        .limit(1)
+    )  # fmt: skip
+    return result.first()
+
+
+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()
+    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(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 connection.engine.url.get_backend_name() != "sqlite":
+        return
+
+    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:
+        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")

src/transcription/db/runtime.py

@@ -0,0 +1,103 @@
+import logging
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from functools import partial
+
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlmodel.ext.asyncio.session import AsyncSession
+from sqlmodel.pool import StaticPool
+
+from ..config import Settings
+from ..config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class DatabaseRuntime:
+    """Database runtime resources owned by app lifespan."""
+
+    engine: AsyncEngine
+    session_factory: async_sessionmaker[AsyncSession]
+
+
+_runtime: ContextVar[DatabaseRuntime | None] = ContextVar("database_runtime", default=None)
+
+
+async def dispose_database_runtime() -> None:
+    """Dispose lifespan-owned async database resources."""
+    runtime = _runtime.get()
+    if runtime is None:
+        return
+    await runtime.engine.dispose()
+    _runtime.set(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)
+    engine_factory = partial(
+        create_async_engine,
+        url=database_url,
+        echo=False,
+        pool_pre_ping=True,
+    )
+
+    if database_url.startswith("sqlite"):
+        sqlite_connect_settings = {"check_same_thread": settings.sqlite_check_same_thread}
+        engine_factory = partial(engine_factory, connect_args=sqlite_connect_settings)
+        if ":memory:" in database_url:
+            engine_factory = partial(engine_factory, poolclass=StaticPool)
+
+    return engine_factory()
+
+
+def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime:
+    """Initialize lifespan-owned async DB resources once per process."""
+    runtime = _runtime.get()
+    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)
+    _runtime.set(runtime)
+    logger.debug("Initialized async database runtime for database_url=%s", engine.url)
+    return runtime
+
+
+def get_engine(settings: Settings | None = None) -> AsyncEngine:
+    """Return the current async SQLAlchemy engine."""
+    runtime = _runtime.get() or initialize_database_runtime(settings=settings)
+    return runtime.engine
+
+
+def get_session_factory(settings: Settings | None = None) -> async_sessionmaker[AsyncSession]:
+    """Return the shared async session factory."""
+    runtime = _runtime.get() or initialize_database_runtime(settings=settings)
+    return runtime.session_factory
+
+
+@asynccontextmanager
+async def get_session(
+    *,
+    settings: Settings | None = None,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+) -> AsyncGenerator[AsyncSession]:
+    """Yield a database session and ensure cleanup."""
+    active_session_factory = session_factory or get_session_factory(settings)
+    async with active_session_factory() as session:
+        yield session

src/transcription/errors.py

@@ -17,6 +17,7 @@ class ErrorCategory(StrEnum):
     NOT_FOUND = "not_found_error"
     CONFLICT = "conflict_error"
     EXTERNAL_PROVIDER = "external_provider_error"
+    PROCESSING = "processing_error"
     INFRA_TRANSIENT = "infrastructure_transient_error"
     INFRA_PERSISTENT = "infrastructure_persistent_error"
     INTERNAL_UNEXPECTED = "internal_unexpected_error"
@@ -81,7 +82,4 @@ def classify_unexpected_error(exc: Exception, *, operation: str) -> AppError:
 
 def format_error_detail(error: AppError) -> str:
     """Return a compact persisted failure string for transcript.error_detail."""
-    return (
+    return f"[{error.category.value}] {error.message} | suggestion={error.suggestion} | error_id={error.error_id}"
-        f"[{error.category.value}] {error.message} | "
-        f"suggestion={error.suggestion} | error_id={error.error_id}"
-    )

src/transcription/providers/base.py

@@ -30,3 +30,4 @@ class TranscriptionProvider(Protocol):
 
     def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
         """Transcribe the provided image according to the prompt text."""
+        ...

src/transcription/services/__init__.py

@@ -1,25 +1,19 @@
 """Service layer exports."""
 
-from transcription.services.transcription import DEFAULT_PROMPT_FILE
+from dataclasses import dataclass
-from transcription.services.transcription import PromptLoadError
+from dataclasses import field
-from transcription.services.transcription import TranscriptionError
-from transcription.services.transcription import load_image_payload
-from transcription.services.transcription import load_prompt_text
-from transcription.services.transcription import transcribe_document_image
-from transcription.services.upload import SUPPORTED_UPLOAD_EXTENSIONS
-from transcription.services.upload import UploadError
-from transcription.services.upload import UploadJobResult
-from transcription.services.upload import create_upload_job
 
-__all__ = [
+from .documents import DocumentService
-    "DEFAULT_PROMPT_FILE",
+from .jobs import JobService
-    "SUPPORTED_UPLOAD_EXTENSIONS",
+from .transcription import TranscriptionService
-    "PromptLoadError",
+
-    "TranscriptionError",
+__all__ = ["DocumentService", "JobService", "ServiceBundle", "TranscriptionService"]
-    "UploadError",
+
-    "UploadJobResult",
+
-    "create_upload_job",
+@dataclass(frozen=True, slots=True)
-    "load_image_payload",
+class ServiceBundle:
-    "load_prompt_text",
+    """Container for all service instances."""
-    "transcribe_document_image",
+
-]
+    documents: DocumentService = field(default_factory=DocumentService)
+    jobs: JobService = field(default_factory=JobService)
+    transcriptions: TranscriptionService = field(default_factory=TranscriptionService)

src/transcription/services/base.py

@@ -0,0 +1,38 @@
+import asyncio
+from abc import ABC
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..config import Settings
+from ..config import get_settings
+from ..db.runtime import get_session_factory
+
+
+class ServiceBase(ABC):
+    """Thin service class for managing documents in the database."""
+
+    settings: Settings
+    session_factory: async_sessionmaker[AsyncSession]
+    queue: asyncio.Queue
+
+    def __init__(
+        self,
+        session_factory: async_sessionmaker[AsyncSession] | None = None,
+        queue: asyncio.Queue | None = None,
+    ):
+        self.settings = get_settings()
+        self.session_factory = session_factory or get_session_factory()
+        self.queue = queue or asyncio.Queue()
+
+    @asynccontextmanager
+    async def _session_scope(self, session: AsyncSession | None = None):
+        """Provide a transactional scope around a series of operations."""
+        if session is not None:
+            # Reuse the provided session if one is passed in
+            yield session
+        else:
+            # Otherwise, create a new session for this scope
+            async with self.session_factory() as new_session:
+                yield new_session

src/transcription/services/documents.py

@@ -0,0 +1,197 @@
+import logging
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from uuid import UUID
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import selectinload
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..config import Settings
+from ..config import get_settings
+from ..errors import AppError
+from ..errors import ErrorCategory
+from ..models import Document
+from ..models import Job
+from ..models import JobStatus
+from .base import ServiceBase
+from .store import store_file
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentError(AppError):
+    """Raised when document operations fail."""
+
+
+class MissingImageError(DocumentError):
+    """Raised when a required image is missing."""
+
+
+class UploadError(DocumentError):
+    """Raised when uploaded content cannot be persisted safely."""
+
+
+class DocumentAlreadyExistsError(DocumentError):
+    """Raised when a document with the same filename already exists in the database."""
+
+
+@dataclass(frozen=True)
+class UploadJobResult:
+    """Summary of created upload records."""
+
+    document_id: UUID
+    job_id: UUID
+    stored_path: Path
+    original_filename: str
+
+
+class DocumentService(ServiceBase):
+    """Thin service class for managing documents in the database."""
+
+    #
+    # CRUD Operations
+    #
+
+    async def create_document(self, document: Document) -> Document:
+        """Create a new document in the database."""
+        async with self.session_factory() as session:
+            session.add(document)
+            try:
+                await session.commit()
+            except IntegrityError as exc:
+                raise DocumentAlreadyExistsError(
+                    f"Document with id {document.id} already exists",
+                    category=ErrorCategory.VALIDATION,
+                    suggestion="Rename the file and try again.",
+                ) from exc
+            await session.refresh(document)
+            return document
+
+    async def read_document(self, document_id: UUID) -> Document:
+        """Read an existing document from the database.
+
+        The selectinload option is used to eagerly load related jobs for the document.
+        """
+        async with self.session_factory() as session:
+            document = await session.get(
+                Document,
+                document_id,
+                options=(selectinload(Document.jobs),),  # pyright: ignore[reportArgumentType]
+            )
+        if document is None:
+            raise DocumentError(
+                f"Document with id {document_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Re-upload the source document and retry.",
+            )
+        elif not Path(document.file_path).exists():
+            raise MissingImageError(
+                f"Document with id {document_id} is missing its image file in {self.settings.upload_dir:!s}",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Re-upload the source document and retry.",
+            )
+        return document
+
+    async def update_document(self, document: Document) -> Document:
+        """Update an existing document in the database."""
+        async with self.session_factory() as session:
+            await session.merge(document)
+            await session.commit()
+            await session.refresh(document)
+            return document
+
+    async def delete_document(self, document: Document) -> None:
+        """Delete a document from the database."""
+        async with self.session_factory() as session:
+            await session.delete(document)
+            await session.commit()
+
+    # Query Operations
+
+    async def query_documents(self, *, filename: str | None = None) -> Sequence[Document]:
+        """Query documents from the database based on provided filters."""
+        async with self.session_factory() as session:
+            query = select(Document)
+            if filename is not None:
+                query = query.where(Document.filename == filename)
+            result = await session.exec(query)
+            return result.all()
+
+    async def list_documents(self) -> Sequence[Document]:
+        """List all documents in the database."""
+        async with self.session_factory() as session:
+            result = await session.exec(select(Document))
+            return result.all()
+
+
+async def create_upload_job(
+    *,
+    filename: str,
+    file_bytes: bytes,
+    session: AsyncSession,
+    settings: Settings | None = None,
+) -> UploadJobResult:
+    """Create upload-backed document and queued job records."""
+    runtime_settings = settings or get_settings()
+    stored_path = store_file(
+        filename=filename,
+        file_bytes=file_bytes,
+        settings=runtime_settings,
+    )
+    try:
+        document, job = await _create_upload_records(
+            session=session,
+            original_filename=filename,
+            stored_path=stored_path,
+        )
+    except Exception as exc:
+        _best_effort_delete(stored_path)
+        raise UploadError(
+            "Failed to create upload database records",
+            category=ErrorCategory.INFRA_TRANSIENT,
+            suggestion="Retry upload. If this keeps happening, verify database availability.",
+            retriable=True,
+        ) from exc
+
+    logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
+    return UploadJobResult(
+        document_id=document.id,
+        job_id=job.id,
+        stored_path=stored_path,
+        original_filename=Path(filename).name,
+    )
+
+
+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)
+    await session.flush()
+
+    job = Job(
+        document_id=document.id,
+        status=JobStatus.QUEUED,
+    )
+    session.add(job)
+    await session.commit()
+    await session.refresh(document)
+    await session.refresh(job)
+    return document, job
+
+
+def _best_effort_delete(path: Path) -> None:
+    try:
+        if path.exists():
+            path.unlink()
+    except OSError:
+        logger.warning("Failed to clean up upload file after DB error: %s", path)

src/transcription/services/jobs.py

@@ -0,0 +1,100 @@
+from collections.abc import Sequence
+from uuid import UUID
+
+from sqlalchemy.orm import selectinload
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..models import Job
+from ..models import JobStatus
+from .base import ServiceBase
+
+
+class JobService(ServiceBase):
+    """Thin service class for managing jobs in the database."""
+
+    #
+    # CRUD Operations
+    #
+
+    async def create_job(self, job: Job, session: AsyncSession | None = None) -> Job:
+        """Create a new job in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(job)
+            await _session.commit()
+            await _session.refresh(job)
+            return job
+
+    async def read_job(self, job_id: UUID, session: AsyncSession | None = None) -> Job:
+        """Read an existing job from the database.
+
+        The selectinload option is used to eagerly load the related document for the job, which makes the full Document
+        model object available in the return Job object.
+        """
+        async with self._session_scope(session) as _session:
+            job = await _session.get(
+                Job,
+                job_id,
+                # Makes the full Document model object available in the return Job object
+                options=(selectinload(Job.document),),  # pyright: ignore[reportArgumentType]
+            )
+        if job is None:
+            raise ValueError(f"Job with id {job_id} not found")
+        return job
+
+    async def update_job(self, job: Job, session: AsyncSession | None = None) -> Job:
+        """Update an existing job in the database."""
+        async with self._session_scope(session) as _session:
+            await _session.merge(job)
+            await _session.commit()
+            await _session.refresh(job)
+            return job
+
+    async def delete_job(self, job: Job, session: AsyncSession | None = None) -> None:
+        """Delete a job from the database."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(job)
+            await _session.commit()
+
+    # Query Operations
+
+    async def query_jobs(
+        self,
+        *,
+        status: JobStatus | None = None,
+        filename: str | None = None,
+    ) -> Sequence[Job]:
+        """Query jobs from the database based on provided filters."""
+        async with self._session_scope(None) as session:
+            query = select(Job)
+            if status is not None:
+                query = query.where(Job.status == status)
+            if filename is not None:
+                query = query.where(Job.document.filename == filename)
+            return (await session.exec(query)).all()
+
+    async def list_jobs(self, session: AsyncSession | None = None, *, load_docs: bool = False) -> Sequence[Job]:
+        """List all jobs in the database."""
+        async with self._session_scope(session) as _session:
+            query = select(Job)
+            if load_docs:
+                query = query.options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+            return (await _session.exec(query)).all()
+
+    # Other Operations
+
+    async def mark_job_status(
+        self,
+        job_id: UUID,
+        status: JobStatus,
+        session: AsyncSession | None = None,
+    ) -> Job:
+        """Mark a job with a new status."""
+        async with self._session_scope(session) as _session:
+            job = await _session.get(Job, job_id)
+            if job is None:
+                raise ValueError(f"Job with id {job_id} not found")
+            job.status = status
+            await _session.commit()
+            await _session.refresh(job)
+            return job

src/transcription/services/store.py

@@ -0,0 +1,70 @@
+import logging
+from pathlib import Path
+from uuid import uuid4
+
+from ..config import Settings
+from ..config import get_settings
+from ..errors import AppError
+from ..errors import ErrorCategory
+
+logger = logging.getLogger(__name__)
+
+SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
+
+
+class UploadError(AppError):
+    """Raised when uploaded content cannot be persisted safely."""
+
+
+def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
+    """Persist an uploaded file to the configured upload directory."""
+    runtime_settings = settings or get_settings()
+    _validate_upload(filename=filename, file_bytes=file_bytes)
+
+    upload_dir = runtime_settings.upload_dir
+    upload_dir.mkdir(parents=True, exist_ok=True)
+
+    stored_name = _build_stored_filename(filename)
+    stored_path = upload_dir / stored_name
+
+    try:
+        stored_path.write_bytes(file_bytes)
+    except OSError as exc:
+        raise UploadError(
+            "Failed to persist upload file",
+            category=ErrorCategory.INFRA_PERSISTENT,
+            suggestion="Check upload directory permissions and available disk space, then retry.",
+        ) from exc
+
+    logger.info("Stored uploaded file: %s", stored_path)
+    return stored_path
+
+
+def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
+    if not file_bytes:
+        raise UploadError(
+            "Upload payload is empty",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Select a non-empty file and try again.",
+        )
+
+    safe_name = Path(filename).name
+    if not safe_name:
+        raise UploadError(
+            "Upload filename is required",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Choose a file with a valid filename and retry.",
+        )
+
+    suffix = Path(safe_name).suffix.lower()
+    if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
+        raise UploadError(
+            f"Unsupported upload extension: {suffix}",
+            category=ErrorCategory.USER_INPUT,
+            suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
+        )
+
+
+def _build_stored_filename(filename: str) -> str:
+    safe_name = Path(filename).name
+    return f"{uuid4()}_{safe_name}"

src/transcription/services/transcription.py

@@ -4,12 +4,19 @@ from __future__ import annotations
 
 import logging
 import mimetypes
+from contextlib import contextmanager
 from pathlib import Path
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.orm import selectinload
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.config import Settings
 from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
+from transcription.models import Transcript
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
 from transcription.providers import ProviderResponseError
@@ -17,6 +24,8 @@ from transcription.providers import TranscriptionProvider
 from transcription.providers import TranscriptionResult
 from transcription.providers import get_transcription_provider
 
+from .base import ServiceBase
+
 logger = logging.getLogger(__name__)
 
 DEFAULT_PROMPT_FILE = "transcribe_document.md"
@@ -31,6 +40,75 @@ class TranscriptionError(AppError):
     """Raised when transcription execution fails."""
 
 
+class TranscriptionNotFoundError(TranscriptionError):
+    """Raised when a transcription is not found in the database."""
+
+
+class TranscriptionService(ServiceBase):
+    """Service class for managing transcription operations.
+
+    This is the top-level service that composes functionality from the other services."""
+
+    provider: TranscriptionProvider
+
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None):
+        super().__init__(session_factory=session_factory)
+        self.provider = get_transcription_provider(settings=self.settings)
+
+    async def create_transcript(self, transcript: Transcript) -> Transcript:
+        """Create a new transcript in the database."""
+        async with self.session_factory() as session:
+            session.add(transcript)
+            await session.commit()
+            await session.refresh(transcript)
+            return transcript
+
+    async def read_transcript(self, transcript_id: UUID) -> Transcript:
+        """Read an existing transcript from the database."""
+        async with self.session_factory() as session:
+            transcript = await session.get(
+                Transcript,
+                transcript_id,
+                # Makes the full Job model object available in the return Transcript object
+                options=(selectinload(Transcript.job),),  # pyright: ignore[reportArgumentType]
+            )
+        if transcript is None:
+            raise TranscriptionNotFoundError(
+                f"Transcript with id {transcript_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Verify the transcript id and retry.",
+            )
+        return transcript
+
+    async def update_transcript(self, transcript: Transcript) -> Transcript:
+        """Update an existing transcript in the database."""
+        async with self.session_factory() as session:
+            await session.merge(transcript)
+            await session.commit()
+            await session.refresh(transcript)
+            return transcript
+
+    async def delete_transcript(self, transcript: Transcript) -> None:
+        """Delete a transcript from the database."""
+        async with self.session_factory() as session:
+            await session.delete(transcript)
+            await session.commit()
+
+    async def transcribe_document(
+        self,
+        image_path: str | Path,
+        *,
+        prompt_name: str = DEFAULT_PROMPT_FILE,
+    ) -> TranscriptionResult:
+        """Transcribe a local image using the configured prompt and provider."""
+        return transcribe_document_image(
+            image_path=image_path,
+            prompt_name=prompt_name,
+            settings=self.settings,
+            provider=self.provider,
+        )
+
+
 def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:
     """Load and validate prompt text from PROMPT_DIR."""
     runtime_settings = settings or get_settings()
@@ -102,12 +180,21 @@ def transcribe_document_image(
     adapter = provider or get_transcription_provider(settings=runtime_settings)
     logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
 
-    try:
+    with handle_transcription_errors():
         result = adapter.transcribe(
             prompt_text=prompt_text,
             image_bytes=image_bytes,
             mime_type=mime_type,
         )
+    logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
+    return result
+
+
+@contextmanager
+def handle_transcription_errors():
+    """Context manager to handle transcription errors."""
+    try:
+        yield
     except ProviderAuthError as exc:
         raise TranscriptionError(
             "Provider authentication failed",
@@ -128,6 +215,3 @@ def transcribe_document_image(
             suggestion="Retry the transcription from jobs. If repeated, check provider availability.",
             retriable=True,
         ) from exc
-
-    logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
-    return result

src/transcription/services/upload.py

@@ -1,158 +0,0 @@
-"""Upload service for storing files and creating queued transcription jobs."""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass
-from pathlib import Path
-from uuid import UUID
-from uuid import uuid4
-
-from sqlmodel.ext.asyncio.session import AsyncSession
-
-from transcription.config import Settings
-from transcription.config import get_settings
-from transcription.db import get_session
-from transcription.errors import AppError
-from transcription.errors import ErrorCategory
-from transcription.models import Document
-from transcription.models import Job
-from transcription.models import JobStatus
-
-logger = logging.getLogger(__name__)
-
-SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
-
-
-class UploadError(AppError):
-    """Raised when uploaded content cannot be persisted safely."""
-
-
-@dataclass(frozen=True)
-class UploadJobResult:
-    """Summary of created upload records."""
-
-    document_id: UUID
-    job_id: UUID
-    stored_path: Path
-    original_filename: str
-
-
-async def create_upload_job(
-    *,
-    filename: str,
-    file_bytes: bytes,
-    session: AsyncSession | None = None,
-    settings: Settings | None = None,
-) -> UploadJobResult:
-    """Persist an uploaded file and create document/job records."""
-    runtime_settings = settings or get_settings()
-    _validate_upload(filename=filename, file_bytes=file_bytes)
-
-    upload_dir = runtime_settings.upload_dir
-    upload_dir.mkdir(parents=True, exist_ok=True)
-
-    stored_name = _build_stored_filename(filename)
-    stored_path = upload_dir / stored_name
-
-    try:
-        stored_path.write_bytes(file_bytes)
-    except OSError as exc:
-        raise UploadError(
-            "Failed to persist upload file",
-            category=ErrorCategory.INFRA_PERSISTENT,
-            suggestion="Check upload directory permissions and available disk space, then retry.",
-        ) from exc
-
-    try:
-        if session is not None:
-            document, job = await _create_upload_records(
-                session=session,
-                original_filename=filename,
-                stored_path=stored_path,
-            )
-        else:
-            async with get_session() as local_session:
-                document, job = await _create_upload_records(
-                    session=local_session,
-                    original_filename=filename,
-                    stored_path=stored_path,
-                )
-    except Exception as exc:
-        _best_effort_delete(stored_path)
-        raise UploadError(
-            "Failed to create upload database records",
-            category=ErrorCategory.INFRA_TRANSIENT,
-            suggestion="Retry upload. If this keeps happening, verify database availability.",
-            retriable=True,
-        ) from exc
-
-    logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
-    return UploadJobResult(
-        document_id=document.id,
-        job_id=job.id,
-        stored_path=stored_path,
-        original_filename=Path(filename).name,
-    )
-
-
-def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
-    if not file_bytes:
-        raise UploadError(
-            "Upload payload is empty",
-            category=ErrorCategory.VALIDATION,
-            suggestion="Select a non-empty file and try again.",
-        )
-
-    safe_name = Path(filename).name
-    if not safe_name:
-        raise UploadError(
-            "Upload filename is required",
-            category=ErrorCategory.VALIDATION,
-            suggestion="Choose a file with a valid filename and retry.",
-        )
-
-    suffix = Path(safe_name).suffix.lower()
-    if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
-        raise UploadError(
-            f"Unsupported upload extension: {suffix}",
-            category=ErrorCategory.USER_INPUT,
-            suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
-        )
-
-
-def _build_stored_filename(filename: str) -> str:
-    safe_name = Path(filename).name
-    return f"{uuid4()}_{safe_name}"
-
-
-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)
-    await session.flush()
-
-    job = Job(
-        document_id=document.id,
-        status=JobStatus.QUEUED,
-    )
-    session.add(job)
-    await session.commit()
-    await session.refresh(document)
-    await session.refresh(job)
-    return document, job
-
-
-def _best_effort_delete(path: Path) -> None:
-    try:
-        if path.exists():
-            path.unlink()
-    except OSError:
-        logger.warning("Failed to clean up upload file after DB error: %s", path)

src/transcription/ui/components/upload.py

@@ -0,0 +1,64 @@
+"""Reusable upload widget for document submission."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable
+from collections.abc import Callable
+
+from nicegui import ui
+from nicegui.binding import bindable_dataclass
+from nicegui.events import UploadEventArguments
+
+from transcription.services.documents import DocumentService
+from transcription.services.documents import UploadError
+from transcription.services.documents import UploadJobResult
+from transcription.ui.components.error_presenter import show_error
+from transcription.ui.components.error_presenter import summarize_error
+
+type UploadSubmitter = Callable[[str, bytes], Awaitable[UploadJobResult]]
+
+
+@bindable_dataclass
+class UploadWidgetState:
+    """Simple state container for upload feedback."""
+
+    loading: bool = False
+    message: str = ""
+
+
+def render_upload_widget(*, service: DocumentService) -> None:
+    """Render upload controls and common status/error handling."""
+    state = UploadWidgetState()
+    status_label = ui.label("Upload a document to start transcription.")
+    status_label.bind_text(state, "message")
+
+    async def on_upload(event: UploadEventArguments) -> None:
+        if state.loading:
+            ui.notify("Upload already in progress. Please wait.", type="warning")
+            return
+
+        state.loading = True
+        status_label.text = "Uploading..."
+        try:
+            payload = await event.file.read()
+            result = await service.upload_file(filename=event.file.name, file_bytes=payload)
+            job_id = result.job_id
+            state.message = f"Created job {job_id}" if job_id is not None else "Upload complete"
+            status_label.text = state.message
+            ui.notify(state.message, type="positive")
+        except UploadError as exc:
+            state.message = summarize_error(exc, operation="upload.submit")
+            status_label.text = f"Upload failed: {state.message}"
+            show_error(exc, title="Upload failed", operation="upload.submit")
+        except Exception as exc:  # noqa: BLE001
+            state.message = summarize_error(exc, operation="upload.submit")
+            status_label.text = f"Upload failed: {state.message}"
+            show_error(exc, title="Upload failed", operation="upload.submit")
+        finally:
+            state.loading = False
+
+    ui.upload(
+        on_upload=on_upload,
+        auto_upload=True,
+        label="Select document file",
+    ).props('accept=".jpg,.jpeg,.png,.tif,.tiff,.pdf"')

src/transcription/ui/pages/upload_page.py

@@ -2,73 +2,20 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
-
 from nicegui import ui
-from nicegui.events import UploadEventArguments
 
-from transcription.services.upload import UploadError
+from transcription.services.documents import DocumentService
-from transcription.services.upload import UploadJobResult
+from transcription.ui.components.upload import render_upload_widget
-from transcription.services.upload import create_upload_job
-from transcription.ui.components.error_presenter import show_error
-from transcription.ui.components.error_presenter import summarize_error
-
-
-@dataclass
-class UploadPageState:
-    """Simple state container for upload page feedback."""
-
-    loading: bool = False
-    message: str = ""
-
-
-def accepted_upload_types() -> str:
-    """Return accepted file type string for upload input."""
-    return ".jpg,.jpeg,.png,.tif,.tiff,.pdf"
-
-
-async def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult:
-    """Create an upload job from incoming file data."""
-    return await create_upload_job(filename=filename, file_bytes=file_bytes)
 
 
 def register_page() -> None:
     """Register the upload page route."""
 
-    @ui.page("/")
+    service = DocumentService()
+
+    @ui.page("/upload", title="Upload Document")
     def upload_page() -> None:
-        state = UploadPageState()
+        render_upload_widget(service=service)
-        status_label = ui.label("Upload a document to start transcription.")
-
-        async def on_upload(event: UploadEventArguments) -> None:
-            if state.loading:
-                ui.notify("Upload already in progress. Please wait.", type="warning")
-                return
-
-            state.loading = True
-            status_label.text = "Uploading..."
-            try:
-                payload = await event.file.read()
-                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")
-            except UploadError as exc:
-                state.message = summarize_error(exc, operation="upload.submit")
-                status_label.text = f"Upload failed: {state.message}"
-                show_error(exc, title="Upload failed", operation="upload.submit")
-            except Exception as exc:  # noqa: BLE001
-                state.message = summarize_error(exc, operation="upload.submit")
-                status_label.text = f"Upload failed: {state.message}"
-                show_error(exc, title="Upload failed", operation="upload.submit")
-            finally:
-                state.loading = False
-
-        ui.upload(
-            on_upload=on_upload,
-            auto_upload=True,
-            label="Select document file",
-        ).props(f"accept={accepted_upload_types()}")
 
         with ui.row():
             ui.link("View jobs", "/jobs")

src/transcription/worker.py

@@ -4,12 +4,16 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from contextlib import contextmanager
+from contextlib import suppress
 from datetime import UTC
 from datetime import datetime
-from threading import Event
+from uuid import UUID
 
-from pydantic import ValidationError
 from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.orm import selectinload
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -20,15 +24,95 @@ from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 from transcription.errors import classify_unexpected_error
 from transcription.errors import format_error_detail
-from transcription.models import Document
 from transcription.models import Job
 from transcription.models import JobStatus
 from transcription.models import Transcript
+from transcription.services.transcription import TranscriptionError
 from transcription.services.transcription import transcribe_document_image
 
+from .services import ServiceBundle
+from .services.jobs import JobService
+
 logger = logging.getLogger(__name__)
 
 
+async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Event):
+    """Main worker loop that consumes jobs from the queue and processes them.
+
+    The queue is for Job UUIDs, and the corresponding documents should already have been uploaded.
+    """
+    service = JobService()
+    while not stop_event.is_set():
+        with handle_worker_exceptions():
+            async with _get_queue_item(queue) as job_id:
+                job = await service.read_job(job_id)
+                asyncio.create_task(process_job(job=job, services=ServiceBundle()))
+
+
+@contextmanager
+def handle_worker_exceptions():
+    """Context manager to log and suppress exceptions in the worker loop."""
+    try:
+        yield
+    except Exception as exc:
+        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.loop")
+        logger.exception(
+            "Worker loop exception error_id=%s category=%s",
+            error.error_id,
+            error.category.value,
+        )
+
+
+@asynccontextmanager
+async def _get_queue_item(queue: asyncio.Queue[UUID]) -> AsyncGenerator[UUID]:
+    """Context manager to enqueue a job and ensure it is marked done."""
+    yield await queue.get()
+    queue.task_done()
+
+
+async def process_job(job: Job, services: ServiceBundle) -> None:
+    """Process a single job using the service bundle."""
+    match job.status:
+        case JobStatus.QUEUED:
+            job.status = JobStatus.PROCESSING
+            try:
+                await services.transcriptions.transcribe_image(job.document.file_path)
+                job.status = JobStatus.TRANSCRIBED
+            except TranscriptionError as exc:
+                job.status = JobStatus.FAILED
+                job.error_message = str(exc)
+            await services.jobs.update_job(job)
+
+
+async def run_worker_loop(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    stop_event: asyncio.Event | None = None,
+    wake_queue: asyncio.Queue[None] | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> None:
+    """Run worker loop until stop_event is set.
+
+    If wake_queue is provided, queue activity wakes the loop immediately while
+    timeout-based wakeups preserve current polling behavior.
+    """
+    while True:
+        if stop_event is not None and stop_event.is_set():
+            logger.info("Worker stop event received")
+            return
+
+        if wake_queue is not None:
+            with suppress(TimeoutError):
+                await asyncio.wait_for(wake_queue.get(), timeout=poll_interval_seconds)
+
+        processed_any = False
+        while await process_next_queued_job(session_factory=session_factory):
+            processed_any = True
+
+        if wake_queue is None and not processed_any:
+            await asyncio.sleep(poll_interval_seconds)
+
+
 async def process_next_queued_job(
     *,
     session: AsyncSession | None = None,
@@ -45,7 +129,13 @@ async def process_next_queued_job(
 
 
 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()
+    query = (
+        select(Job)
+        .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+        .where(Job.status == JobStatus.QUEUED)
+        .order_by(Job.created_at)  # pyright: ignore[reportArgumentType]
+    )
+    job = (await session.exec(query)).first()
 
     if job is None:
         return False
@@ -57,14 +147,14 @@ async def _process_next_queued_job(*, session: AsyncSession) -> bool:
     await session.commit()
     await session.refresh(job)
 
-    document = await session.get(Document, job.document_id)
+    document = job.document
     if document is None:
         error = AppError(
             "Document not found",
             category=ErrorCategory.NOT_FOUND,
             suggestion="Re-upload the source document and retry processing.",
         )
-        _finalize_failed_job(session=session, job=job, error=error)
+        await _finalize_failed_job(session=session, job=job, error=error)
         logger.error(
             "Job failed operation=worker.process_job job_id=%s error_id=%s category=%s",
             job.id,
@@ -74,7 +164,8 @@ async def _process_next_queued_job(*, session: AsyncSession) -> bool:
         return True
 
     try:
-        result = transcribe_document_image(document.file_path)
+        # Provider SDK calls are synchronous and should not block the event loop.
+        result = await asyncio.to_thread(transcribe_document_image, document.file_path)
         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)
@@ -88,11 +179,12 @@ async def _process_next_queued_job(*, session: AsyncSession) -> bool:
         )
     except Exception as exc:
         error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job")
-        settings = _get_worker_settings()
+        settings = get_settings()
         if _should_retry(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 retried operation=worker.process_job "
+                "job_id=%s document_id=%s retry_count=%s error_id=%s category=%s",
                 job.id,
                 document.id,
                 job.retry_count,
@@ -127,13 +219,6 @@ async def _upsert_transcript(
     return transcript
 
 
-def _get_worker_settings() -> Settings:
-    try:
-        return get_settings()
-    except ValidationError:
-        return Settings(openrouter_api_key="test-key")
-
-
 def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool:
     return error.retriable and job.retry_count < settings.worker_max_retries
 
@@ -155,36 +240,3 @@ async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppErr
     job.updated_at = datetime.now(UTC)
     session.add(job)
     await session.commit()
-
-
-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 = await process_next_queued_job(session_factory=session_factory)
-        if not processed:
-            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,
-        )
-    )

tests/conftest.py

@@ -5,18 +5,45 @@ isolated, fast, and leave no artifacts on disk.
 """
 
 import pytest
-from sqlmodel import Session, SQLModel, create_engine
+import pytest_asyncio
+from sqlmodel import Session
+from sqlmodel import SQLModel
+from sqlmodel import create_engine
 from sqlmodel.pool import StaticPool
 
+from transcription.config import Settings
+from transcription.config import get_settings
+from transcription.db.operations import create_all
+from transcription.db.runtime import dispose_database_runtime
+from transcription.db.runtime import get_engine
+from transcription.db.runtime import get_session
+
 
 @pytest.fixture
 def session():
-    """Provide a clean database session for each test."""
+    """Provide a clean synchronous database session for sync tests."""
     engine = create_engine(
         "sqlite://",
         connect_args={"check_same_thread": False},
         poolclass=StaticPool,
     )
     SQLModel.metadata.create_all(engine)
-    with Session(engine) as session:
+    with Session(engine) as sync_session:
-        yield session
+        yield sync_session
+
+
+@pytest_asyncio.fixture
+async def default_settings():
+    """Provide default settings for tests."""
+    settings = get_settings(database_url="sqlite:///:memory:")
+    await create_all(engine=get_engine(settings=settings))
+    return settings
+
+
+@pytest_asyncio.fixture
+async def async_session(default_settings: Settings):
+    """Provide a clean asynchronous database session for async tests."""
+    async with get_session(settings=default_settings) as async_session:
+        yield async_session
+
+    await dispose_database_runtime()

tests/services/test_job_service.py

@@ -0,0 +1,88 @@
+from uuid import uuid4
+
+import pytest
+
+from transcription.models import Document
+from transcription.models import Job
+from transcription.services.documents import DocumentService
+from transcription.services.jobs import JobService
+from transcription.services.jobs import JobStatus
+
+
+class TestJobService:
+    class TestBasicCRUD:
+        @pytest.mark.asyncio
+        async def test_create_job(self, job_service: JobService):
+            """Test creating a job."""
+
+            def fake_job_factory():
+                return Job(document_id=uuid4())
+
+            await job_service.create_job(job=fake_job_factory())
+
+            async with job_service._session_scope() as session:
+                for _ in range(10):
+                    await job_service.create_job(job=fake_job_factory(), session=session)
+
+        @pytest.mark.asyncio
+        async def test_backpropagation(self, job_service: JobService, document_service: DocumentService):
+            """Test that creating a job backpropagates to the related document."""
+            doc_id = uuid4()
+            document = Document(
+                id=doc_id,
+                filename="test.txt",
+                file_path="/path/to/test.txt",
+            )
+            await document_service.create_document(document=document)
+            job = Job(document_id=doc_id)
+            await job_service.create_job(job=job)
+
+            read_job = await job_service.read_job(job_id=job.id)
+            assert isinstance(read_job.document, Document)
+            assert read_job.document.id == document.id
+
+        @pytest.mark.asyncio
+        async def test_reading_job(self, job_service: JobService):
+            """Test reading a job."""
+            uuid = uuid4()
+            await job_service.create_job(job=Job(id=uuid, document_id=uuid4()))
+            job = await job_service.read_job(job_id=uuid)
+            assert job.id == uuid
+
+        @pytest.mark.asyncio
+        async def test_updating_job(self, job_service: JobService):
+            """Test updating a job."""
+            uuid = uuid4()
+            job = Job(id=uuid, document_id=uuid4())
+            async with job_service._session_scope() as session:
+                await job_service.create_job(job=job, session=session)
+                job.status = JobStatus.PROCESSING
+                await job_service.update_job(job=job, session=session)
+                read_job = await job_service.read_job(job_id=uuid, session=session)
+            assert read_job == job
+
+        @pytest.mark.asyncio
+        async def test_deleting_job(self, job_service: JobService):
+            """Test deleting a job."""
+
+    class TestServiceMethods:
+        @pytest.mark.asyncio
+        async def test_query_jobs(self, job_service: JobService):
+            """Test querying jobs."""
+            await job_service.create_job(job=Job(document_id=uuid4(), status=JobStatus.PROCESSING))
+            result = await job_service.query_jobs(status=JobStatus.PROCESSING)
+            jobs = {str(job.id).split("-")[0]: job.status for job in result}
+            assert len(jobs) == 1
+
+        @pytest.mark.asyncio
+        async def test_list_jobs(self, job_service: JobService):
+            """Test listing jobs."""
+            n = 5
+            for _ in range(n):
+                await job_service.create_job(job=Job(document_id=uuid4()))
+            jobs = await job_service.list_jobs()
+            assert len(jobs) == n
+
+        @pytest.mark.asyncio
+        async def test_mark_job_status(self, job_service: JobService):
+            """Test marking a job with a new status."""

tests/services/test_service_base.py

@@ -0,0 +1,36 @@
+import pytest
+
+
+class TestServiceBase:
+    class TestInitialization:
+        def test_initializes_with_defaults(self):
+            """Test initialization with default session factory and queue."""
+
+        def test_initializes_with_custom_session_factory(self):
+            """Test initialization with a provided session factory."""
+
+        def test_initializes_with_custom_queue(self):
+            """Test initialization with a provided queue."""
+
+    class TestSessionScope:
+        @pytest.mark.asyncio
+        async def test_uses_provided_session(self):
+            """Test that session scope reuses a provided session."""
+
+        @pytest.mark.asyncio
+        async def test_creates_new_session_when_none_provided(self):
+            """Test that session scope creates a new session when none is provided."""
+
+    class TestContextManagerBehavior:
+        @pytest.mark.asyncio
+        async def test_yields_session(self):
+            """Test that session scope yields a usable session object."""
+
+        @pytest.mark.asyncio
+        async def test_multiple_operations(self):
+            """Test multiple operations within a single session scope."""
+
+    class TestEdgeCases:
+        @pytest.mark.asyncio
+        async def test_handles_exception_propagation(self):
+            """Test exception propagation behavior inside session scope."""