types

.dockerignore

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

.github/instructions/services.instructions.md

@@ -0,0 +1,77 @@
+---
+description: Follow these guidelines when editing the services
+applyTo: 'src/transcription/services/*.py'
+---
+
+# Services
+
+## Structure
+
+- Project core data models defined in [models](../../src/transcription/models.py)
+- 1 service class per data model
+- Only services directly interact with the database, and only through async methods
+- Services are completely independent of one another. Any operation that needs to use more than a single service, which is most of them, needs to have a separate orchestration function.
+
+## Error Handling
+
+- Service-specific errors defined at the top of the respective module and inherit from `AppError`
+- Use a context manager for large `try/except` blocks like in [transcription](../../src/transcription/services/transcription.py)
+
+## Checklist
+
+- [ ] Uses `ServiceBase` for common logic
+- [ ] CRUD methods created at the top
+- [ ] Session kwarg for `AsyncSession` to pass in a session object to each method
+- [ ] Services use `self._session_scope` in their methods to pass the session thru.
+    - Multiple operations on the same object(s) require sharing a session between all the methods used.
+
+## CRUD Methods
+
+- Create, read, update, and delete, created in that order
+- Name format `<operation>_<model >`, for example `create_document` or `update_job`
+- All services must define these 4 methods first, and in that order
+
+## Transaction Finalization
+
+When a service method accepts an optional `session` kwarg, write methods must use `self._finalize` to finalize the transaction properly according to whether or not they are sharing a session.
+
+- If `session` is `None`: the method owns the transaction and should `commit()`.
+- If `session` is provided: the method must **not** commit; it should `flush()` so IDs and FK values are available to the caller's transaction.
+- Use `refresh()` on returned ORM objects when the caller needs DB-populated values (defaults, triggers, merged state).
+
+Recommended helper behavior:
+
+- Inputs: active session object, original `session` arg (or a boolean ownership flag), and an optional list of objects to refresh.
+- Logic: `commit` when service-owned session, `flush` when caller-owned session, then refresh requested objects.
+
+This keeps orchestration functions atomic: they can pass one shared session across multiple services and commit exactly once at the workflow boundary.
+
+## Workflow Transaction Boundaries
+
+For multi-step job lifecycles (for example queued transcription jobs), orchestration functions must use explicit transaction phases.
+
+Required boundary model:
+
+- **Transaction A (claim):** transition `JobStatus.QUEUED -> JobStatus.PROCESSING` and commit immediately.
+- Perform provider/network work **outside** database transactions.
+- **Transaction B (terminal success):** write transcript content and set `JobStatus.TRANSCRIBED` in the same shared-session commit.
+- **Transaction B (terminal failure):** write transcript error detail and set `JobStatus.FAILED` in the same shared-session commit.
+- **Transaction C (retry path):** write transcript error detail, increment retry count, and set `JobStatus.QUEUED` in one shared-session commit.
+
+Atomicity rules:
+
+- Never commit transcript updates separately from the paired terminal/retry job status change.
+- Terminal state (`TRANSCRIBED` or `FAILED`) and transcript row changes must succeed or roll back together.
+- Retry persistence (`QUEUED` + retry increment + error detail) must succeed or roll back together.
+
+Separation of concerns:
+
+- Worker modules should stay lightweight and delegate lifecycle transitions to service/workflow orchestration functions.
+- In `workflows.py`, `process_queued_job` should own one complete attempt lifecycle: `QUEUED -> PROCESSING -> TRANSCRIBED|FAILED`.
+- In `workflows.py`, `advance_job` should coordinate broader status progression around attempts (for example retry scheduling from `FAILED -> QUEUED`).
+- Services should expose session-aware write helpers (flush on caller-owned session) so orchestration controls commit boundaries.
+- Backoff/sleep behavior must run outside transactional scopes.
+
+# Service Composition
+
+Some operations, like uploading a picutre, require modifications to multiple tables, which can be done by composing methods from the service object into a separate function.

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,18 +12,29 @@ 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",
 ]
 
-[project.optional-dependencies]
+
+[dependency-groups]
 dev = [
     "pytest>=8.0",
     "pytest-asyncio>=0.25",
+    "httpx2>=2.5.0",
+    "ipykernel>=7.3.0",
+    "ipywidgets>=8.1.8",
+    "pre-commit>=4.6.0",
+    "rich>=15.0.0",
+    "ruff>=0.15.20",
+    "ty>=0.0.54",
 ]
 
 [tool.pytest.ini_options]

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 threading import Thread
+from contextlib import suppress
 
 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 = Event()
-    worker_thread = Thread(
-        target=run_worker_loop,
-        kwargs={
-            "session_factory": session_factory,
-            "stop_event": stop_event,
-            "poll_interval_seconds": 1.0,
-        },
-        daemon=True,
+    stop_event = asyncio.Event()
+    wake_queue: asyncio.Queue[None] = asyncio.Queue()
+    worker_task = asyncio.create_task(
+        run_worker_loop(
+            session_factory=app.state.runtime.session_factory,
+            stop_event=stop_event,
+            wake_queue=wake_queue,
+            poll_interval_seconds=1.0,
+        )
     )
-    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:
-        worker_thread.join(timeout=2.0)
+    if wake_queue is not None:
+        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.db_engine = runtime.engine
-    app.state.db_session_factory = runtime.session_factory
+    app.state.services = ServiceBundle()
+    app.state.runtime = initialize_database_runtime(settings=settings)
 
     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 cleanup_database()
+        await _stop_worker(app)
+        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
 
@@ -74,7 +75,7 @@ LOGGING_CONFIG: dict[str, object] = {
     "disable_existing_loggers": False,
     "formatters": {
         "standard": {
-            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
+            "format": "%(asctime)s %(levelname)-8s | %(message)s",
             "datefmt": "%Y-%m-%d %H:%M:%S",
         }
     },

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 (
-        f"[{error.category.value}] {error.message} | "
-        f"suggestion={error.suggestion} | error_id={error.error_id}"
-    )
+    return f"[{error.category.value}] {error.message} | suggestion={error.suggestion} | error_id={error.error_id}"

src/transcription/models.py

@@ -29,9 +29,7 @@ class Document(SQLModel, table=True):
     id: UUID = Field(default_factory=uuid4, primary_key=True)
     filename: str
     file_path: str
-    uploaded_at: datetime = Field(
-        default_factory=lambda: datetime.now(UTC),
-    )
+    uploaded_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     # --- relationships ---
     jobs: list["Job"] = Relationship(back_populates="document")
@@ -44,28 +42,34 @@ class Job(SQLModel, table=True):
     document_id: UUID = Field(foreign_key="document.id")
     status: JobStatus = Field(default=JobStatus.QUEUED)
     retry_count: int = Field(default=0, ge=0)
-    created_at: datetime = Field(
-        default_factory=lambda: datetime.now(UTC),
-    )
-    updated_at: datetime = Field(
-        default_factory=lambda: datetime.now(UTC),
-    )
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     # --- relationships ---
     document: Document = Relationship(back_populates="jobs")
     transcript: Optional["Transcript"] = Relationship(back_populates="job")
 
+    @property
+    def filename(self) -> str:
+        """Return the filename of the associated document."""
+        return self.document.filename if self.document else "unknown"
+
 
 class Transcript(SQLModel, table=True):
     """The output of a transcription job."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
     job_id: UUID = Field(foreign_key="job.id", unique=True)
+    """ID for the associated job. There's a 1-1 relationship bewteen transcripts and jobs."""
+    provider: str
+    """Name of the transcription provider used to generate this transcript."""
+    prompt_name: str
+    """Name of the prompt used to generate this transcript."""
     text: str | None = None
+    """The transcribed text. This may be None if the job failed or is still in progress."""
     error_detail: str | None = None
-    created_at: datetime = Field(
-        default_factory=lambda: datetime.now(UTC),
-    )
+    """Details of any error that occurred during transcription."""
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     # --- relationships ---
     job: Job = Relationship(back_populates="transcript")

src/transcription/providers/base.py

@@ -2,6 +2,9 @@
 
 from dataclasses import dataclass
 from typing import Protocol
+from uuid import UUID
+
+from ..models import Transcript
 
 
 class ProviderError(RuntimeError):
@@ -22,11 +25,22 @@ class TranscriptionResult:
 
     text: str
     provider: str
+    prompt_name: str
     model: str
 
+    def to_transcript(self, job_id: UUID) -> Transcript:
+        """Convert a TranscriptionResult to a Transcript model instance."""
+        return Transcript(
+            job_id=job_id,
+            provider=self.provider,
+            prompt_name=self.prompt_name,
+            text=self.text,
+        )
+
 
 class TranscriptionProvider(Protocol):
     """Contract every transcription provider adapter must satisfy."""
 
-    def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+    async def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
         """Transcribe the provided image according to the prompt text."""
+        ...

src/transcription/providers/openrouter.py

@@ -6,8 +6,10 @@ import base64
 import logging
 from dataclasses import dataclass
 from typing import Any
+from typing import cast
 
 from openrouter import OpenRouter
+from openrouter.components.chatmessages import ChatMessagesTypedDict
 
 from transcription.config import Settings
 from transcription.config import get_settings
@@ -44,12 +46,12 @@ class OpenRouterTranscriptionProvider:
         """Return the resolved OpenRouter model slug."""
         return self._model
 
-    def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+    async def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
         """Send prompt + image to OpenRouter and return normalized text output."""
         request = self._build_request(prompt_text=prompt_text, image_bytes=image_bytes, mime_type=mime_type)
         try:
-            response = self._client.chat.send(
-                messages=request.messages,
+            response = await self._client.chat.send_async(
+                messages=cast(list[ChatMessagesTypedDict], request.messages),
                 model=request.model,
                 http_referer=request.http_referer,
                 x_open_router_title=request.x_open_router_title,
@@ -63,7 +65,7 @@ class OpenRouterTranscriptionProvider:
         text = self._extract_text(response)
         model = self._get_optional_attr(response, "model") or self.model
         logger.info("OpenRouter transcription completed using model=%s", model)
-        return TranscriptionResult(text=text, provider="openrouter", model=model)
+        return TranscriptionResult(text=text, provider="openrouter", prompt_name="", model=model)
 
     def _build_request(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> OpenRouterRequest:
         image_b64 = base64.b64encode(image_bytes).decode("ascii")

src/transcription/services/__init__.py

@@ -1,25 +1,19 @@
 """Service layer exports."""
 
-from transcription.services.transcription import DEFAULT_PROMPT_FILE
-from transcription.services.transcription import PromptLoadError
-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
+from dataclasses import dataclass
+from dataclasses import field
 
-__all__ = [
-    "DEFAULT_PROMPT_FILE",
-    "SUPPORTED_UPLOAD_EXTENSIONS",
-    "PromptLoadError",
-    "TranscriptionError",
-    "UploadError",
-    "UploadJobResult",
-    "create_upload_job",
-    "load_image_payload",
-    "load_prompt_text",
-    "transcribe_document_image",
-]
+from .documents import DocumentService
+from .jobs import JobService
+from .transcription import TranscriptionService
+
+__all__ = ["DocumentService", "JobService", "ServiceBundle", "TranscriptionService"]
+
+
+@dataclass(frozen=True, slots=True)
+class ServiceBundle:
+    """Container for all service instances."""
+
+    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,60 @@
+import asyncio
+from abc import ABC
+from collections.abc import Sequence
+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
+
+    async def _finalize(
+        self,
+        *,
+        session: AsyncSession,
+        caller_session: AsyncSession | None,
+        refresh: Sequence[object] = (),
+    ) -> None:
+        """Finalize a write based on transaction ownership.
+
+        Service-owned sessions commit immediately. Caller-owned sessions flush so
+        orchestration code can commit once at a larger transaction boundary.
+        """
+        should_commit = caller_session is None
+        if should_commit:
+            await session.commit()
+        else:
+            await session.flush()
+
+        for obj in refresh:
+            await session.refresh(obj)

src/transcription/services/documents.py

@@ -0,0 +1,127 @@
+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 ..errors import AppError
+from ..errors import ErrorCategory
+from ..models import Document
+from .base import ServiceBase
+
+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,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Document:
+        """Create a new document in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(document)
+            try:
+                await self._finalize(session=_session, caller_session=session, refresh=(document,))
+            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
+            return document
+
+    async def read_document(self, document_id: UUID, *, session: AsyncSession | None = None) -> 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_scope(session) 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, *, session: AsyncSession | None = None) -> Document:
+        """Update an existing document in the database."""
+        async with self._session_scope(session) as _session:
+            merged = await _session.merge(document)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def delete_document(self, document: Document, *, session: AsyncSession | None = None) -> None:
+        """Delete a document from the database."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(document)
+            await self._finalize(session=_session, caller_session=session)
+
+    # Query Operations
+
+    async def query_documents(
+        self, *, filename: str | None = None, session: AsyncSession | None = None
+    ) -> Sequence[Document]:
+        """Query documents from the database based on provided filters."""
+        async with self._session_scope(session) 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, *, session: AsyncSession | None = None) -> Sequence[Document]:
+        """List all documents in the database."""
+        async with self._session_scope(session) as _session:
+            result = await _session.exec(select(Document))
+            return result.all()

src/transcription/services/jobs.py

@@ -0,0 +1,146 @@
+from collections.abc import Sequence
+from datetime import UTC
+from datetime import datetime
+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 self._finalize(session=_session, caller_session=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 related document is always eagerly loaded so callers can safely
+        access ``job.document`` in async contexts without triggering lazy-load IO.
+        """
+        async with self._session_scope(session) as _session:
+            query = (
+                select(Job)
+                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .where(Job.id == job_id)
+                .execution_options(populate_existing=True)
+            )
+            job = (await _session.exec(query)).first()
+        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:
+            merged = await _session.merge(job)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    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 self._finalize(session=_session, caller_session=session)
+
+    # Query Operations
+
+    async def query_jobs(
+        self,
+        *,
+        status: JobStatus | None = None,
+        filename: str | None = None,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Job]:
+        """Query jobs from the database based on provided filters."""
+        async with self._session_scope(session) as _session:
+            query = select(Job).options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+            if status is not None:
+                query = query.where(Job.status == status)
+            if filename is not None:
+                query = query.where(Job.document.filename == filename)
+            result = await _session.exec(query)
+            return result.all()
+
+    async def list_jobs(
+        self,
+        *,
+        load_docs: bool = False,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Job]:
+        """List all jobs in the database with eagerly loaded documents."""
+        _ = load_docs
+        async with self._session_scope(session) as _session:
+            query = select(Job).options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+            result = await _session.exec(query)
+            return result.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."""
+        return await self.update_job_state(job_id=job_id, status=status, session=session)
+
+    async def update_job_state(
+        self,
+        *,
+        job_id: UUID,
+        status: JobStatus,
+        retry_count_increment: int = 0,
+        session: AsyncSession | None = None,
+    ) -> Job:
+        """Update a job's lifecycle fields.
+
+        When ``session`` is provided, this method flushes so callers can commit
+        once at an orchestration boundary.
+        """
+        async with self._session_scope(session) as _session:
+            query = (
+                select(Job)
+                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .where(Job.id == job_id)
+                .execution_options(populate_existing=True)
+            )
+            job = (await _session.exec(query)).first()
+            if job is None:
+                raise ValueError(f"Job with id {job_id} not found")
+            job.status = status
+            if retry_count_increment:
+                job.retry_count += retry_count_increment
+            job.updated_at = datetime.now(UTC)
+            await self._finalize(session=_session, caller_session=session, refresh=(job,))
+            return job
+
+    async def read_next_queued_job(
+        self,
+        *,
+        session: AsyncSession | None = None,
+    ) -> Job | None:
+        """Read the next queued job ordered by creation time."""
+        async with self._session_scope(session) as _session:
+            query = (
+                select(Job)
+                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .where(Job.status == JobStatus.QUEUED)
+                .order_by(Job.created_at)  # pyright: ignore[reportArgumentType]
+            )
+            return (await _session.exec(query)).first()

src/transcription/services/store.py

@@ -1,23 +1,19 @@
-"""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
+
+from ..models import Document
+from ..models import Job
+from .documents import UploadJobResult
 
 logger = logging.getLogger(__name__)
 
@@ -28,24 +24,75 @@ 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,
+    session: AsyncSession,
     settings: Settings | None = None,
 ) -> UploadJobResult:
-    """Persist an uploaded file and create document/job records."""
+    """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)
+    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)
+
+
+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)
 
@@ -64,36 +111,8 @@ async def create_upload_job(
             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,
-    )
+    logger.info("Stored uploaded file: %s", stored_path)
+    return stored_path
 
 
 def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
@@ -124,35 +143,3 @@ def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
 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/services/transcription.py

@@ -4,12 +4,20 @@ 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 import select
+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 +25,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 +41,131 @@ 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, *, session: AsyncSession | None = None) -> Transcript:
+        """Create a new transcript in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(transcript)
+            await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
+            return transcript
+
+    async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Transcript:
+        """Read an existing transcript from the database."""
+        async with self._session_scope(session) 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, *, session: AsyncSession | None = None) -> Transcript:
+        """Update an existing transcript in the database."""
+        async with self._session_scope(session) as _session:
+            merged = await _session.merge(transcript)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def delete_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> None:
+        """Delete a transcript from the database."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(transcript)
+            await self._finalize(session=_session, caller_session=session)
+
+    async def transcribe_document(
+        self,
+        image_path: str | Path,
+        job_id: UUID,
+        *,
+        prompt_name: str = DEFAULT_PROMPT_FILE,
+        session: AsyncSession | None = None,
+    ):
+        """Transcribe a local image using the configured prompt and provider."""
+        result = await transcribe_document_image(
+            image_path=image_path,
+            prompt_name=prompt_name,
+            settings=self.settings,
+            provider=self.provider,
+        )
+        await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session)
+
+    async def upsert_transcript_by_job(
+        self,
+        *,
+        job_id: UUID,
+        text: str | None,
+        error_detail: str | None,
+        provider: str | None = None,
+        prompt_name: str = DEFAULT_PROMPT_FILE,
+        session: AsyncSession | None = None,
+    ) -> Transcript:
+        """Create or update a transcript for a job id."""
+        async with self._session_scope(session) as _session:
+            transcript = (await _session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
+            if transcript is None:
+                transcript = Transcript(
+                    job_id=job_id,
+                    provider=provider or self.settings.provider.value,
+                    prompt_name=prompt_name,
+                )
+
+            transcript.text = text
+            transcript.error_detail = error_detail
+            if provider is not None:
+                transcript.provider = provider
+            transcript.prompt_name = prompt_name
+
+            _session.add(transcript)
+            await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
+            return transcript
+
+
+async def transcribe_document_image(
+    image_path: str | Path,
+    *,
+    prompt_name: str = DEFAULT_PROMPT_FILE,
+    settings: Settings | None = None,
+    provider: TranscriptionProvider | None = None,
+) -> TranscriptionResult:
+    """Transcribe a local image using the configured prompt and provider."""
+    runtime_settings = settings or get_settings()
+    prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings)
+    image_bytes, mime_type = load_image_payload(image_path)
+
+    adapter = provider or get_transcription_provider(settings=runtime_settings)
+    logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
+
+    with handle_transcription_errors():
+        result = await 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
+
+
 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()
@@ -87,27 +222,11 @@ def load_image_payload(image_path: str | Path) -> tuple[bytes, str]:
     return path.read_bytes(), mime_type
 
 
-def transcribe_document_image(
-    image_path: str | Path,
-    *,
-    prompt_name: str = DEFAULT_PROMPT_FILE,
-    settings: Settings | None = None,
-    provider: TranscriptionProvider | None = None,
-) -> TranscriptionResult:
-    """Transcribe a local image using the configured prompt and provider."""
-    runtime_settings = settings or get_settings()
-    prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings)
-    image_bytes, mime_type = load_image_payload(image_path)
-
-    adapter = provider or get_transcription_provider(settings=runtime_settings)
-    logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
-
+@contextmanager
+def handle_transcription_errors():
+    """Context manager to handle transcription errors."""
     try:
-        result = adapter.transcribe(
-            prompt_text=prompt_text,
-            image_bytes=image_bytes,
-            mime_type=mime_type,
-        )
+        yield
     except ProviderAuthError as exc:
         raise TranscriptionError(
             "Provider authentication failed",
@@ -128,6 +247,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/workflows.py

@@ -0,0 +1,241 @@
+import asyncio
+import logging
+
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..config import Settings
+from ..config import get_settings
+from ..errors import AppError
+from ..errors import classify_unexpected_error
+from ..errors import format_error_detail
+from ..models import Job
+from ..models import JobStatus
+from ..providers import TranscriptionResult
+from . import ServiceBundle
+from .transcription import DEFAULT_PROMPT_FILE
+from .transcription import transcribe_document_image
+
+logger = logging.getLogger(__name__)
+
+
+async def advance_job(
+    job: Job,
+    services: ServiceBundle,
+    settings: Settings | None = None,
+    session: AsyncSession | None = None,
+) -> Job | None:
+    """Advance a single job by lifecycle status."""
+    settings = settings or get_settings()
+    match job.status:
+        case JobStatus.QUEUED:
+            return await process_queued_job(job=job, services=services, session=session)
+        case JobStatus.FAILED:
+            if job.retry_count < settings.worker_max_retries:
+                return await services.jobs.update_job_state(
+                    job_id=job.id,
+                    status=JobStatus.QUEUED,
+                    retry_count_increment=1,
+                    session=session,
+                )
+            else:
+                logger.error(f"Job {job.id} has failed and reached max retries.")
+                return
+        case _:
+            return
+
+
+async def process_queued_job(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    session: AsyncSession | None = None,
+) -> Job | None:
+    """Process one complete transcription attempt for a queued job."""
+    if job.status != JobStatus.QUEUED:
+        logger.warning(f"Job {job.id} is not queued. Current status: {job.status}")
+        return
+
+    # Transaction A: claim job for processing.
+    if session is None:
+        job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING)
+    else:
+        # If we're sharing the session, need to make sure setting the Job to PROCESSING is committed before we start
+        # the transcription, otherwise other workers may see the job as still QUEUED and try to process it.
+        job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session)
+        await session.commit()
+
+    document = job.document
+    assert document is not None, (
+        f"Job {job.id} has no associated document or the document failed to be loaded by the job service."
+    )
+
+    try:
+        result = await transcribe_document_image(document.file_path)
+        job = await _finalize_transcribed(job=job, services=services, result=result, session=session)
+        logger.info(
+            "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s",
+            job.id,
+            document.id,
+            result.provider,
+        )
+    except Exception as exc:  # noqa: BLE001
+        match exc:
+            case AppError() as error:
+                pass
+            case _:
+                error = classify_unexpected_error(exc, operation="worker.process_job")
+
+        job = await _finalize_failed(job=job, services=services, error=error, session=session)
+        logger.error(
+            "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
+            job.id,
+            document.id,
+            error.error_id,
+            error.category.value,
+        )
+    return job
+
+
+async def process_next_queued_job(
+    *,
+    services: ServiceBundle,
+    settings: Settings | None = None,
+    session: AsyncSession | None = None,
+) -> bool:
+    """Process the next queued job if one exists."""
+    job = await services.jobs.read_next_queued_job(session=session)
+    if job is None:
+        return False
+
+    await advance_job(job=job, services=services, settings=settings, session=session)
+    return True
+
+
+async def _finalize_transcribed(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    result: TranscriptionResult,
+    session: AsyncSession | None = None,
+) -> Job:
+    """Transaction B: transcript + TRANSCRIBED in one commit."""
+    if session is None:
+        async with services.jobs._session_scope() as local_session:
+            await services.transcriptions.upsert_transcript_by_job(
+                job_id=job.id,
+                text=result.text,
+                error_detail=None,
+                provider=result.provider,
+                prompt_name=result.prompt_name,
+                session=local_session,
+            )
+            updated_job = await services.jobs.mark_job_status(
+                job.id,
+                JobStatus.TRANSCRIBED,
+                session=local_session,
+            )
+            await local_session.commit()
+            return updated_job
+
+    await services.transcriptions.upsert_transcript_by_job(
+        job_id=job.id,
+        text=result.text,
+        error_detail=None,
+        provider=result.provider,
+        prompt_name=result.prompt_name,
+        session=session,
+    )
+    updated_job = await services.jobs.mark_job_status(
+        job.id,
+        JobStatus.TRANSCRIBED,
+        session=session,
+    )
+    await session.commit()
+    return updated_job
+
+
+async def _finalize_retry(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    error: AppError,
+    settings: Settings,
+    session: AsyncSession | None = None,
+) -> Job:
+    """Transaction C: transcript error + QUEUED + retry increment in one commit."""
+    if session is None:
+        async with services.jobs._session_scope() as local_session:
+            await services.transcriptions.upsert_transcript_by_job(
+                job_id=job.id,
+                text=None,
+                error_detail=format_error_detail(error),
+                prompt_name=DEFAULT_PROMPT_FILE,
+                session=local_session,
+            )
+            updated_job = await services.jobs.update_job_state(
+                job_id=job.id,
+                status=JobStatus.QUEUED,
+                retry_count_increment=1,
+                session=local_session,
+            )
+            await local_session.commit()
+    else:
+        await services.transcriptions.upsert_transcript_by_job(
+            job_id=job.id,
+            text=None,
+            error_detail=format_error_detail(error),
+            prompt_name=DEFAULT_PROMPT_FILE,
+            session=session,
+        )
+        updated_job = await services.jobs.update_job_state(
+            job_id=job.id,
+            status=JobStatus.QUEUED,
+            retry_count_increment=1,
+            session=session,
+        )
+        await session.commit()
+
+    if settings.worker_retry_backoff_seconds > 0:
+        await asyncio.sleep(settings.worker_retry_backoff_seconds)
+    return updated_job
+
+
+async def _finalize_failed(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    error: AppError,
+    session: AsyncSession | None = None,
+) -> Job:
+    """Transaction B: transcript error + FAILED in one commit."""
+    if session is None:
+        async with services.jobs._session_scope() as local_session:
+            await services.transcriptions.upsert_transcript_by_job(
+                job_id=job.id,
+                text=None,
+                error_detail=format_error_detail(error),
+                prompt_name=DEFAULT_PROMPT_FILE,
+                session=local_session,
+            )
+            updated_job = await services.jobs.mark_job_status(
+                job.id,
+                JobStatus.FAILED,
+                session=local_session,
+            )
+            await local_session.commit()
+            return updated_job
+
+    await services.transcriptions.upsert_transcript_by_job(
+        job_id=job.id,
+        text=None,
+        error_detail=format_error_detail(error),
+        prompt_name=DEFAULT_PROMPT_FILE,
+        session=session,
+    )
+    updated_job = await services.jobs.mark_job_status(
+        job.id,
+        JobStatus.FAILED,
+        session=session,
+    )
+    await session.commit()
+    return updated_job

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.upload import UploadJobResult
-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)
+from transcription.services.documents import DocumentService
+from transcription.ui.components.upload import render_upload_widget
 
 
 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()
-        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()}")
+        render_upload_widget(service=service)
 
         with ui.row():
             ui.link("View jobs", "/jobs")

src/transcription/worker.py

@@ -4,31 +4,92 @@ from __future__ import annotations
 
 import asyncio
 import logging
-from datetime import UTC
-from datetime import datetime
-from threading import Event
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from contextlib import contextmanager
+from contextlib import suppress
+from uuid import UUID
 
-from pydantic import ValidationError
 from sqlalchemy.ext.asyncio import async_sessionmaker
-from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from transcription.config import Settings
-from transcription.config import get_settings
 from transcription.db import get_session
 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 transcribe_document_image
+
+from .services import ServiceBundle
+from .services.documents import DocumentService
+from .services.jobs import JobService
+from .services.transcription import TranscriptionService
+from .services.workflows import advance_job
+from .services.workflows import process_next_queued_job as process_next_queued_job_workflow
 
 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(advance_job(job=job, services=ServiceBundle()))
+
+
+@contextmanager
+def handle_worker_exceptions(operation: str = "worker.loop"):
+    """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=operation)
+        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 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,
@@ -38,153 +99,17 @@ async def process_next_queued_job(
 
     Returns True when a job was processed, False when no queued job exists.
     """
+    if session_factory is None:
+        services = ServiceBundle()
+    else:
+        services = ServiceBundle(
+            documents=DocumentService(session_factory=session_factory),
+            jobs=JobService(session_factory=session_factory),
+            transcriptions=TranscriptionService(session_factory=session_factory),
+        )
+
     if session is None:
         async with get_session(session_factory=session_factory) as local_session:
-            return await _process_next_queued_job(session=local_session)
-    return await _process_next_queued_job(session=session)
+            return await process_next_queued_job_workflow(services=services, session=local_session)
 
-
-async def _process_next_queued_job(*, session: AsyncSession) -> bool:
-    job = (await session.exec(select(Job).where(Job.status == JobStatus.QUEUED).order_by(Job.created_at))).first()
-
-    if job is None:
-        return False
-
-    logger.info("Picked queued job operation=worker.pick job_id=%s", job.id)
-    job.status = JobStatus.PROCESSING
-    job.updated_at = datetime.now(UTC)
-    session.add(job)
-    await session.commit()
-    await session.refresh(job)
-
-    document = await session.get(Document, job.document_id)
-    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)
-        logger.error(
-            "Job failed operation=worker.process_job job_id=%s error_id=%s category=%s",
-            job.id,
-            error.error_id,
-            error.category.value,
-        )
-        return True
-
-    try:
-        result = 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)
-        session.add(job)
-        await session.commit()
-        logger.info(
-            "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s",
-            job.id,
-            document.id,
-            result.provider,
-        )
-    except Exception as exc:
-        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job")
-        settings = _get_worker_settings()
-        if _should_retry(job=job, error=error, settings=settings):
-            await _requeue_for_retry(session=session, job=job, error=error, settings=settings)
-            logger.warning(
-                "Job retried operation=worker.process_job job_id=%s document_id=%s retry_count=%s error_id=%s category=%s",
-                job.id,
-                document.id,
-                job.retry_count,
-                error.error_id,
-                error.category.value,
-            )
-        else:
-            await _finalize_failed_job(session=session, job=job, error=error)
-            logger.exception(
-                "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
-                job.id,
-                document.id,
-                error.error_id,
-                error.category.value,
-            )
-
-    return True
-
-
-async def _upsert_transcript(
-    *, session: AsyncSession, job_id, text: str | None, error_detail: str | None
-) -> Transcript:
-    transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
-    if transcript is None:
-        transcript = Transcript(job_id=job_id)
-
-    transcript.text = text
-    transcript.error_detail = error_detail
-    session.add(transcript)
-    await session.commit()
-    await session.refresh(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
-
-
-async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None:
-    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
-    job.retry_count += 1
-    job.status = JobStatus.QUEUED
-    job.updated_at = datetime.now(UTC)
-    session.add(job)
-    await session.commit()
-    if settings.worker_retry_backoff_seconds > 0:
-        await asyncio.sleep(settings.worker_retry_backoff_seconds)
-
-
-async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None:
-    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
-    job.status = JobStatus.FAILED
-    job.updated_at = datetime.now(UTC)
-    session.add(job)
-    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,
-        )
-    )
+    return await process_next_queued_job_workflow(services=services, session=session)

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:
-        yield session
+    with Session(engine) as sync_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."""

tests/services/test_transcription.py

@@ -1,137 +0,0 @@
-"""Tests for transcription.services.transcription."""
-
-from pathlib import Path
-
-import pytest
-
-from transcription.config import Settings
-from transcription.providers.base import ProviderError, TranscriptionResult
-from transcription.services.transcription import (
-    PromptLoadError,
-    TranscriptionError,
-    load_image_payload,
-    load_prompt_text,
-    transcribe_document_image,
-)
-
-
-class _FakeProvider:
-    def __init__(self, *, result: TranscriptionResult | None = None, error: Exception | None = None):
-        self._result = result or TranscriptionResult(
-            text="Transcript output",
-            provider="openrouter",
-            model="test-model",
-        )
-        self._error = error
-        self.calls: list[dict[str, object]] = []
-
-    def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
-        self.calls.append(
-            {
-                "prompt_text": prompt_text,
-                "image_bytes": image_bytes,
-                "mime_type": mime_type,
-            }
-        )
-        if self._error:
-            raise self._error
-        return self._result
-
-
-@pytest.mark.unit
-class TestPromptLoading:
-    """Verify prompt artifact loading and validation."""
-
-    def test_loads_prompt_text_from_prompt_dir(self, tmp_path: Path):
-        """Prompt loader returns canonical prompt text from configured prompt directory."""
-        prompt_dir = tmp_path / "prompts"
-        prompt_dir.mkdir()
-        prompt_file = prompt_dir / "transcribe_document.md"
-        prompt_file.write_text("Prompt body", encoding="utf-8")
-
-        settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
-        text = load_prompt_text(settings=settings)
-
-        assert text == "Prompt body"
-
-    def test_missing_prompt_raises_error(self, tmp_path: Path):
-        """Prompt loader raises PromptLoadError when the file is missing."""
-        prompt_dir = tmp_path / "prompts"
-        prompt_dir.mkdir()
-
-        settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
-        with pytest.raises(PromptLoadError) as exc_info:
-            load_prompt_text(settings=settings)
-
-        assert exc_info.value.category.value == "infrastructure_persistent_error"
-        assert "verify prompt_dir" in exc_info.value.suggestion.lower()
-
-
-@pytest.mark.unit
-class TestImageLoading:
-    """Verify local image payload loading and mime detection."""
-
-    def test_load_image_payload_reads_bytes_and_mime_type(self, tmp_path: Path):
-        """Image loader returns file bytes and a detected MIME type for supported files."""
-        image_path = tmp_path / "sample.png"
-        image_bytes = b"\x89PNG\r\n\x1a\n"
-        image_path.write_bytes(image_bytes)
-
-        loaded_bytes, mime_type = load_image_payload(image_path)
-
-        assert loaded_bytes == image_bytes
-        assert mime_type == "image/png"
-
-    def test_missing_image_raises_error(self, tmp_path: Path):
-        """Image loader raises TranscriptionError when image file does not exist."""
-        missing = tmp_path / "missing.png"
-
-        with pytest.raises(TranscriptionError) as exc_info:
-            load_image_payload(missing)
-
-        assert exc_info.value.category.value == "not_found_error"
-        assert "verify" in exc_info.value.suggestion.lower()
-
-
-@pytest.mark.unit
-class TestTranscriptionService:
-    """Verify service orchestration across prompt, image, and provider calls."""
-
-    def test_transcribe_document_image_calls_provider_once(self, tmp_path: Path):
-        """Service loads prompt and image, then invokes provider exactly once."""
-        prompt_dir = tmp_path / "prompts"
-        prompt_dir.mkdir()
-        (prompt_dir / "transcribe_document.md").write_text("Prompt body", encoding="utf-8")
-
-        image_path = tmp_path / "document.jpg"
-        image_path.write_bytes(b"jpeg-bytes")
-
-        settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
-        provider = _FakeProvider()
-
-        result = transcribe_document_image(image_path, settings=settings, provider=provider)
-
-        assert result.text == "Transcript output"
-        assert len(provider.calls) == 1
-        assert provider.calls[0]["prompt_text"] == "Prompt body"
-        assert provider.calls[0]["image_bytes"] == b"jpeg-bytes"
-        assert provider.calls[0]["mime_type"] == "image/jpeg"
-
-    def test_provider_error_is_wrapped(self, tmp_path: Path):
-        """Service wraps provider failures in TranscriptionError."""
-        prompt_dir = tmp_path / "prompts"
-        prompt_dir.mkdir()
-        (prompt_dir / "transcribe_document.md").write_text("Prompt body", encoding="utf-8")
-
-        image_path = tmp_path / "document.png"
-        image_path.write_bytes(b"png-bytes")
-
-        settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
-        provider = _FakeProvider(error=ProviderError("upstream failure"))
-
-        with pytest.raises(TranscriptionError) as exc_info:
-            transcribe_document_image(image_path, settings=settings, provider=provider)
-
-        assert exc_info.value.category.value == "external_provider_error"
-        assert exc_info.value.retriable is True
-        assert "retry" in exc_info.value.suggestion.lower()

tests/services/test_upload.py

@@ -1,104 +0,0 @@
-"""Tests for transcription.services.upload."""
-
-from pathlib import Path
-
-import pytest
-
-from transcription.config import Settings
-from transcription.models import Document, Job, JobStatus
-from transcription.services.upload import UploadError, create_upload_job
-
-
-@pytest.mark.unit
-class TestUploadValidation:
-    """Verify upload validation behavior."""
-
-    def test_rejects_empty_bytes(self, session, tmp_path: Path):
-        """create_upload_job rejects an empty upload payload."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        with pytest.raises(UploadError) as exc_info:
-            create_upload_job(
-                filename="letter.jpg",
-                file_bytes=b"",
-                session=session,
-                settings=settings,
-            )
-
-        assert exc_info.value.category.value == "validation_error"
-        assert "non-empty" in exc_info.value.suggestion.lower()
-
-    def test_rejects_unsupported_extension(self, session, tmp_path: Path):
-        """create_upload_job rejects unsupported filename extensions."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-        with pytest.raises(UploadError) as exc_info:
-            create_upload_job(
-                filename="notes.txt",
-                file_bytes=b"content",
-                session=session,
-                settings=settings,
-            )
-
-        assert exc_info.value.category.value == "user_input_error"
-        assert "jpg" in exc_info.value.suggestion.lower()
-
-
-@pytest.mark.integration
-class TestUploadPersistence:
-    """Verify upload file and record persistence behavior."""
-
-    def test_writes_file_and_creates_records(self, session, tmp_path: Path):
-        """create_upload_job writes file and creates document/job records."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-
-        result = create_upload_job(
-            filename="letter.jpg",
-            file_bytes=b"image-bytes",
-            session=session,
-            settings=settings,
-        )
-
-        assert result.stored_path.exists()
-        assert result.stored_path.read_bytes() == b"image-bytes"
-
-        document = session.get(Document, result.document_id)
-        job = session.get(Job, result.job_id)
-        assert document is not None
-        assert job is not None
-        assert document.filename == "letter.jpg"
-        assert document.file_path == str(result.stored_path)
-
-    def test_uses_unique_stored_filename(self, session, tmp_path: Path):
-        """create_upload_job stores uploads with unique filenames."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-
-        first = create_upload_job(
-            filename="duplicate.jpg",
-            file_bytes=b"first",
-            session=session,
-            settings=settings,
-        )
-        second = create_upload_job(
-            filename="duplicate.jpg",
-            file_bytes=b"second",
-            session=session,
-            settings=settings,
-        )
-
-        assert first.stored_path != second.stored_path
-        assert first.stored_path.exists()
-        assert second.stored_path.exists()
-
-    def test_sets_job_status_queued(self, session, tmp_path: Path):
-        """create_upload_job persists a job with queued status."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
-
-        result = create_upload_job(
-            filename="queued.pdf",
-            file_bytes=b"%PDF-1.4",
-            session=session,
-            settings=settings,
-        )
-
-        job = session.get(Job, result.job_id)
-        assert job is not None
-        assert job.status == JobStatus.QUEUED

tests/services/test_worker.py

@@ -1,232 +0,0 @@
-"""Tests for transcription.worker."""
-
-from pathlib import Path
-from threading import Event
-
-import pytest
-from sqlmodel import select
-
-from transcription.config import Settings
-from transcription.errors import AppError, ErrorCategory
-from transcription.models import Document, Job, JobStatus, Transcript
-from transcription.providers.base import TranscriptionResult
-from transcription.worker import process_next_queued_job, run_worker_loop
-
-
-def _create_queued_job(session, *, filename: str = "doc.jpg", file_path: str = "uploads/doc.jpg") -> Job:
-    document = Document(filename=filename, file_path=file_path)
-    session.add(document)
-    session.commit()
-    session.refresh(document)
-
-    job = Job(document_id=document.id, status=JobStatus.QUEUED)
-    session.add(job)
-    session.commit()
-    session.refresh(job)
-    return job
-
-
-@pytest.mark.integration
-class TestWorkerQueueBehavior:
-    """Verify worker behavior when selecting queued jobs."""
-
-    def test_returns_false_when_queue_empty(self, session):
-        """process_next_queued_job returns False when there are no queued jobs."""
-        processed = process_next_queued_job(session=session)
-        assert processed is False
-
-
-@pytest.mark.integration
-class TestWorkerSuccessPath:
-    """Verify worker success-path lifecycle transitions and transcript persistence."""
-
-    def test_transitions_processing_to_transcribed(self, session, monkeypatch, tmp_path: Path):
-        """process_next_queued_job transitions queued jobs to transcribed on success."""
-        job = _create_queued_job(session)
-
-        def _fake_transcribe(_path):
-            return TranscriptionResult(text="ok", provider="openrouter", model="test-model")
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
-        )
-
-        processed = process_next_queued_job(session=session)
-        session.refresh(job)
-
-        assert processed is True
-        assert job.status == JobStatus.TRANSCRIBED
-
-    def test_persists_transcript_text_on_success(self, session, monkeypatch, tmp_path: Path):
-        """process_next_queued_job stores transcript text for successful jobs."""
-        job = _create_queued_job(session)
-
-        def _fake_transcribe(_path):
-            return TranscriptionResult(text="Transcript body", provider="openrouter", model="test-model")
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
-        )
-
-        process_next_queued_job(session=session)
-
-        transcript = session.exec(
-            select(Transcript).where(Transcript.job_id == job.id)
-        ).first()
-        assert transcript is not None
-        assert transcript.text == "Transcript body"
-        assert transcript.error_detail is None
-
-
-@pytest.mark.integration
-class TestWorkerFailurePath:
-    """Verify worker failure-path lifecycle transitions and error persistence."""
-
-    def test_sets_failed_and_error_detail_on_failure(self, session, monkeypatch, tmp_path: Path):
-        """process_next_queued_job marks failed and stores error detail on exception."""
-        job = _create_queued_job(session)
-
-        def _fake_transcribe(_path):
-            raise RuntimeError("provider failure")
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
-        )
-
-        processed = process_next_queued_job(session=session)
-        session.refresh(job)
-        transcript = session.exec(
-            select(Transcript).where(Transcript.job_id == job.id)
-        ).first()
-
-        assert processed is True
-        assert job.status == JobStatus.FAILED
-        assert transcript is not None
-        assert transcript.text is None
-        assert "provider failure" in transcript.error_detail
-        assert "[internal_unexpected_error]" in transcript.error_detail
-        assert "error_id=" in transcript.error_detail
-        assert "suggestion=" in transcript.error_detail
-
-    def test_updates_existing_transcript_if_present(self, session, monkeypatch, tmp_path: Path):
-        """process_next_queued_job updates existing transcript instead of duplicating."""
-        job = _create_queued_job(session)
-        existing = Transcript(job_id=job.id, text="old", error_detail=None)
-        session.add(existing)
-        session.commit()
-        session.refresh(existing)
-
-        def _fake_transcribe(_path):
-            raise RuntimeError("provider failure")
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
-        )
-
-        process_next_queued_job(session=session)
-
-        transcripts = session.exec(
-            select(Transcript).where(Transcript.job_id == job.id)
-        ).all()
-        assert len(transcripts) == 1
-        assert transcripts[0].id == existing.id
-        assert transcripts[0].text is None
-        assert "provider failure" in transcripts[0].error_detail
-        assert "[internal_unexpected_error]" in transcripts[0].error_detail
-        assert "error_id=" in transcripts[0].error_detail
-
-
-@pytest.mark.integration
-class TestWorkerRetryBehavior:
-    """Verify worker retry and terminal failure policies."""
-
-    def test_retriable_failure_requeues_until_limit(self, session, monkeypatch, tmp_path: Path):
-        """Retriable failures requeue jobs while retry budget remains."""
-        job = _create_queued_job(session)
-
-        def _fake_transcribe(_path):
-            raise AppError(
-                "temporary upstream outage",
-                category=ErrorCategory.EXTERNAL_PROVIDER,
-                suggestion="Retry from jobs page.",
-                retriable=True,
-            )
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(
-                openrouter_api_key="test-key",
-                prompt_dir=tmp_path,
-                worker_max_retries=1,
-                worker_retry_backoff_seconds=0.0,
-            ),
-        )
-
-        processed = process_next_queued_job(session=session)
-        session.refresh(job)
-
-        assert processed is True
-        assert job.status == JobStatus.QUEUED
-        assert job.retry_count == 1
-
-    def test_retriable_failure_exhaustion_sets_failed(self, session, monkeypatch, tmp_path: Path):
-        """Retriable failures transition to failed when retry budget is exhausted."""
-        job = _create_queued_job(session)
-        job.retry_count = 1
-        session.add(job)
-        session.commit()
-
-        def _fake_transcribe(_path):
-            raise AppError(
-                "temporary upstream outage",
-                category=ErrorCategory.EXTERNAL_PROVIDER,
-                suggestion="Retry from jobs page.",
-                retriable=True,
-            )
-
-        monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
-        monkeypatch.setattr(
-            "transcription.worker.get_settings",
-            lambda: Settings(
-                openrouter_api_key="test-key",
-                prompt_dir=tmp_path,
-                worker_max_retries=1,
-                worker_retry_backoff_seconds=0.0,
-            ),
-        )
-
-        process_next_queued_job(session=session)
-        session.refresh(job)
-
-        assert job.status == JobStatus.FAILED
-        assert job.retry_count == 1
-
-
-@pytest.mark.unit
-class TestWorkerLoopControl:
-    """Verify worker loop start/stop behavior."""
-
-    def test_stops_when_stop_event_is_set(self, monkeypatch):
-        """run_worker_loop exits when a stop event is set."""
-        stop_event = Event()
-        stop_event.set()
-
-        called = {"value": False}
-
-        def _fake_process_next_queued_job(**_kwargs):
-            called["value"] = True
-            return False
-
-        monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
-
-        run_worker_loop(stop_event=stop_event, poll_interval_seconds=0.01)
-        assert called["value"] is False