Merge commit '4ae8e5be4f60059ba611ceea0cf553b009e6a88e' into session-engine

docs/implementation_plan_v2.md

@@ -1,248 +1,57 @@
-# Implementation Plan (Version 2)
+# implementation_plan_v2
 
-This plan defines the path from the V1 baseline to **Version 2 complete**, aligned to the updated multi-image and multi-person relational domain model:
+## Goal
 
-* `Document` acts as a logical parent container for physical artifacts, supporting multi-author and multi-recipient relationships via `DocumentPerson`.
+Replace the current V1 SQLModel schema with the approved V2 schema and make sure every database operation works through the existing async SQLAlchemy/SQLModel session layer.
-* `Source` represents an individual image page within a document, maintaining sequential order (`page_number`), cached active machine output (`raw_transcription`), and inline single user revisions (`revised_text`).
-* `Job` acts as an overarching batch orchestrator for multi-page async processing tasks.
-* `JobSource` records individual point-in-time API executions per image page, storing Pydantic-validated `ai_metadata` and raw REST envelopes (`raw_api_response`).
-* **Pydantic V2** acts as the single source of truth for runtime validation, API payload parsing, and PostgreSQL JSONB serialization.
 
-The objective is to complete the V2 scope with production readiness while keeping non-V2 enhancements out of active delivery.
+Use a fresh database. There will be no migrations, data conversion, legacy compatibility shims, or parallel V1/V2 code paths.
 
----
+## Current Project Impact
 
-## V2 Completion Definition
+- `src/transcription/db/models.py` still defines the V1 `Document`, `Source`, `Job`, and `Revision` tables.
+- The V2 target adds `Person`, `DocumentPerson`, and `JobSource`, moves revisions onto `Source`, and removes the direct `Source.job_id` relationship.
+- The engine, session factory, transaction handling, and PostgreSQL async support already exist and do not need to be rewritten.
+- Async CRUD currently lives in `DocumentService`, `JobService`, `TranscriptionService`, and the upload record helper. Their queries and eager-loading options depend on V1 relationships.
+- Existing tests cover only part of the schema and CRUD surface.
 
-V2 is complete when all of the following are true:
+## Implementation
 
-1. **Functional complete**
+### 1. Update the schema
-* Multi-image and whole-folder uploads assign sequential page numbers to `Source` records under a single `Document`.
-* Batch jobs process pages concurrently using an `asyncio` worker pool with semaphore rate limiting.
-* Partial job failures resolve cleanly to `partial_success`, allowing single-page retries without re-running successful pages.
-* Multi-author and multi-recipient tagging is supported on `Document`.
 
+- Replace the models in `src/transcription/db/models.py` with the approved V2 tables, enums, relationships, foreign keys, constraints, and indexes.
+- Remove `Revision`, `Source.job_id`, and the transcription fields that no longer belong on `Job`.
+- Keep `create_all()` as the schema bootstrap for a fresh database.
+- Delete `_ensure_sqlite_compat_columns()` and all schema patching from `src/transcription/db/operations.py`.
+- Keep the Python models, `docs/schema_v2.md`, and `docs/ddl_v2.sql` consistent.
 
-2. **Data-model complete**
+### 2. Align the async CRUD methods
-* SQLite is fully replaced with PostgreSQL (using `asyncpg` or `psycopg3`).
-* Pydantic V2 models validate all API payloads, database row mappings, and `JSONB` structures.
 
+- Keep the existing `ServiceBase` session and transaction pattern.
+- Update document CRUD to load and manage its ordered `Source` rows and `DocumentPerson` links.
+- Update job CRUD and queue queries to use `JobSource` instead of `Source.job_id`.
+- Add the missing async CRUD operations for `Person`, `Source`, `DocumentPerson`, and `JobSource` using the existing service style. Do not add another repository abstraction.
+- Replace revision CRUD with direct updates to `Source.revised_text` and `Source.date_revised`.
+- Remove the temporary transcript compatibility aliases instead of redirecting them.
+- Update only direct database call sites that construct or query these records; UI and worker feature changes are not part of this work.
 
-3. **Operational complete**
+### 3. Verify the schema and CRUD
-* Concurrency controls, worker pool metrics, and database connections operate safely under batch load.
 
+- Update the schema bootstrap test to expect `person`, `document`, `document_person`, `source`, `job`, and `job_source`, with no `revision` table.
+- Add async create, read, update, delete, list, and filtered-query tests for each entity that exposes those operations.
+- Test relationship loading, page ordering, uniqueness constraints, delete behavior, status values, and `JobSource` JSON fields.
+- Test both service-owned sessions and caller-provided sessions so flush/commit behavior remains correct.
+- Run the focused database and service tests, then the full suite with `uv run pytest`.
 
-4. **Documentation complete**
+## Done When
-* `schema_v2.md`, `DDL_v2.sql`, Pydantic model contracts are updated and consistent.
 
+- A fresh database is created directly from the V2 SQLModel metadata.
+- All async CRUD methods pass against the V2 relationships and fields.
+- No code references `Revision`, `Source.job_id`, removed `Job` transcription fields, or compatibility aliases.
+- The focused tests and full test suite pass.
 
+## Out of Scope
 
----
+- Database migrations or preservation of V1 data
-
+- Legacy compatibility code
-## Phase 1 — Data Contract Stabilization & Pydantic Baseline
+- Database engine or session-layer rewrites
-
+- UI redesign, batch orchestration, worker concurrency, deployment, and operational runbooks
-**Goal:** Lock the PostgreSQL schema, DDL, and Pydantic V2 models before refactoring service logic.
-
-### Tasks
-
-1. Finalize DDL for PostgreSQL native types (`UUID`, `TIMESTAMPTZ`, `JSONB`) and junction tables (`document_person`, `job_source`).
-2. Build core Pydantic V2 schemas (`Person`, `Document`, `Source`, `Job`, `JobSource`, `PageAIMetadata`).
-3. Confirm and document data invariants:
-* `source.raw_transcription` and `job_source.raw_transcription` are immutable machine outputs.
-* `source.revised_text` holds user edits. UI renders `COALESCE(revised_text, raw_transcription)`.
-* Page sequence is strictly ordered by `source.page_number ASC`.
-
-
-4. Freeze V2 job status values (`queued`, `processing`, `completed`, `partial_success`, `failed`) and page execution status values (`pending`, `transcribed`, `failed`).
-
-### Deliverables
-
-* Canonical `docs/schema_v2.md` and `docs/DDL_v2.sql`.
-* Centralized Pydantic validation suite in `models/schemas_v2.py`.
-
-### Exit Criteria
-
-* All database tables, relationships, and JSONB structures have corresponding Pydantic V2 models passing unit validation tests.
-
----
-
-## Phase 2 — Persistence Layer Transition (SQLite to PostgreSQL)
-
-**Goal:** Replace the SQLite storage layer with an asynchronous PostgreSQL driver (`asyncpg` or `psycopg3`).
-
-### Tasks
-
-1. Configure PostgreSQL database connection pooling and environment configuration.
-2. Refactor `services/store.py` / repository layers to execute parameterized async SQL queries (`$1`, `$2`).
-3. Implement JSONB serialization and deserialization helpers using Pydantic's `.model_dump_json()` and `.model_validate()`.
-4. Implement database bootstrap routines for PostgreSQL table creation and index initialization.
-
-### Deliverables
-
-* PostgreSQL-native database connection and query service modules.
-* Integration test suite confirming connection pooling and JSONB CRUD operations.
-
-### Exit Criteria
-
-* All database reads/writes run asynchronously against PostgreSQL with zero remaining SQLite driver dependencies.
-
----
-
-## Phase 3 — Service Layer & `asyncio` Engine Refactor
-
-**Goal:** Implement batch orchestration and parallel single-image API execution.
-
-### Tasks
-
-1. Refactor upload service to process folder/multi-image input:
-* Group files into a single `Document`.
-* Create ordered `Source` rows (`page_number = 1..N`).
-
-
-2. Refactor `services/workflows.py` with `asyncio` worker pools:
-* Use `asyncio.Semaphore` to enforce API provider rate limits.
-* Issue parallel single-image requests to Vision APIs (OpenAI/Claude).
-* Parse API responses directly into Pydantic models (`PageAIMetadata`).
-
-
-3. Update execution tracking:
-* Create a `JobSource` row per page call to record `raw_transcription`, `ai_metadata`, and `raw_api_response`.
-* Update active `source.raw_transcription` upon task completion.
-* Calculate aggregate batch status (`completed`, `partial_success`, `failed`) on the parent `Job`.
-
-
-4. Refactor `services/person.py` and `services/documents.py` to handle multi-person roles via `document_person`.
-
-### Deliverables
-
-* Asynchronous batch execution engine in `services/workflows.py`.
-* Service routines for multi-person tagging and page-level retries.
-
-### Exit Criteria
-
-* Executing a folder upload of 10+ images processes concurrently, populates page-level `JobSource` entries, and handles partial worker errors without crashing the batch.
-
----
-
-## Phase 4 — UI & API Contract Alignment
-
-**Goal:** Update API endpoints and frontend/UI views to render multi-page documents and person roles.
-
-### Tasks
-
-1. Update document and job API endpoints to accept batch file arrays and multi-person ID payloads.
-2. Update UI document views:
-* Render multi-page document transcriptions sequentially by `page_number`.
-* Display author and recipient chips/cards linked from `document_person`.
-
-
-3. Update job detail UI to show page-level execution statuses (`transcribed` vs. `failed`) and provide a "Retry Failed Pages" action for `partial_success` jobs.
-4. Align inline page editing controls to update `source.revised_text` and `source.date_revised`.
-
-### Deliverables
-
-* Refactored API routes and UI components supporting multi-page rendering and person management.
-
-### Exit Criteria
-
-* UI successfully displays multi-page document text, allows per-page human revisions, and shows author/recipient metadata.
-
----
-
-## Phase 5 — Test Suite Realignment & Concurrency Testing
-
-**Goal:** Ensure end-to-end system stability under concurrent async execution and load.
-
-### Tasks
-
-1. Write unit tests for Pydantic models, custom validators, and JSONB conversions.
-2. Write integration tests for async database operations:
-* CRUD for `Document`, `Person`, `DocumentPerson`, `Source`, `Job`, and `JobSource`.
-
-
-3. Write mock-backed async workflow tests:
-* Verify `asyncio.Semaphore` bounds concurrent tasks properly.
-* Validate state transition logic for `completed`, `partial_success`, and `failed` jobs.
-* Confirm retry routines process only targeted `JobSource` records marked as `failed`.
-
-
-4. Re-enable CI quality gates (linting, type checking with Pyright/mypy, pytest).
-
-### Deliverables
-
-* Passing asynchronous test suite covering core workflows, edge cases, and failure recoveries.
-
-### Exit Criteria
-
-* CI pipeline is green with comprehensive coverage across database operations, Pydantic models, and worker queues.
-
----
-
-## Phase 6 — Reliability, Operations, and Release Readiness
-
-**Goal:** Prepare V2 for production deployment and operator management.
-
-### Tasks
-
-1. Verify structured logging includes `job_id`, `document_id`, `source_id`, and `person_id`.
-2. Tune PostgreSQL connection pool limits and `asyncio` concurrency thresholds for production infrastructure.
-3. Update operational documentation:
-* Review and update `docs/schema_v2.md` as needed.
-* Create `docs/runbook_v2.md` detailing PostgreSQL maintenance, JSONB index management, and worker queue monitoring.
-* Create `docs/release_checklist_v2.md` for launch sign-off.
-
-
-
-### Deliverables
-
-* Updated project documentation and operational runbooks.
-* V2 release sign-off checklist.
-
-### Exit Criteria
-
-* All documentation reflects V2 architecture; launch checklist is fully verified.
-
----
-
-## Requirement Traceability Focus
-
-Maintain evidence against these V2 requirement groups:
-
-* **Batch & Multi-Image Pipeline:** Folder ingestion, page ordering, async worker execution.
-* **Database & Persistence:** PostgreSQL, native UUIDs, JSONB execution storage, `asyncpg` pooling.
-* **Validation & Schemas:** Pydantic V2 models for DB rows, API requests, and AI vision responses.
-* **Attribution & Metadata:** Multi-author and multi-recipient tagging, biographical entity management.
-* **Error Recovery:** Partial success states, page-level status flags, isolated retry execution.
-
----
-
-## Scope Discipline Rule (V2 Focus)
-
-* Only tasks required for V2 scope (PostgreSQL, Pydantic V2, folder/async processing, multi-person roles) enter this plan.
-* V3 candidate features (such as side-by-side multi-provider model output comparison) remain strictly in the future backlog.
-* Any schema adjustments during implementation require immediate updates to `DDL_v2.sql`, Pydantic models, and `schema_v2.md`.
-
----
-
-## Technology References
-
-- [FastAPI documentation](https://fastapi.tiangolo.com/)
-- [NiceGUI documentation](https://nicegui.io/documentation)
-- [PostgreSQL documentation](https://www.postgresql.org/docs/)
-- [Python asyncio](https://docs.python.org/3/library/asyncio.html#module-asyncio)
-- [Pydantic Validation](https://pydantic.dev/docs/validation/latest/get-started/)
-- [Pydantic AI](https://pydantic.dev/docs/ai/overview/)
-
-## Related Local References
-
-- [System Overview](index_v2.md)
-- [System Design Intent](intent.md)
-- [Transcription Methodology](transcription_methodology.md)
-- [System Architecture](architecture_v2.md)
-- [System Requirements](requirements_v2.md)
-- [Data model](schema_v2.md)
-- [Error Handling Policy](error_handling_v2.md)
-- Implementation Plan (this document)
- 
-
-

docs/index_v2.md

@@ -19,7 +19,7 @@ Read [architecture_v2.md](architecture_v2.md) first for technical overview and s
 ## Technical Stack
 
 * **Application Web Framework:** FastAPI + NiceGUI
-* **Persistence Engine:** PostgreSQL 13+
+* **Persistence Engine:** PostgreSQL 18+
 * **Data Validation & Schemas:** Pydantic V2
 * **Concurrency & Workers:** Python `asyncio` worker pool with `asyncio.Semaphore`
 * **Vision Providers:** OpenAI (GPT-4o) and Anthropic (Claude 3.5 Sonnet) via native SDKs

docs/ver2/V2 Context for DB Updates.md

@@ -0,0 +1,35 @@
+# AI Coding Assistant Project Briefing & Context
+
+## Project Mission
+This application is a family history archival and transcription platform. Its primary goal is to accept scanned document images (letters, postcards, logbooks, diaries), execute OCR and structured transcription via AI vision models (OpenAI GPT-4o, Anthropic Claude 3.5 Sonnet), and manage historical metadata (authors, recipients, dates, and locations).
+
+---
+
+## Technical Stack & Architecture
+* **Database:** PostgreSQL 13+ with native `UUID` (`gen_random_uuid()`) and `JSONB` columns.
+* **Backend Runtime / Concurrency:** Python utilizing `asyncio` for concurrent HTTP API calls to AI providers, with strict rate-limiting via `asyncio.Semaphore`.
+* **Validation & Types:** Python with **Pydantic** model definitions. Incoming AI responses must be parsed and validated with Pydantic models *before* database insertion.
+* **ORM / Database Access:** SQLModel and SQLAlchemy, using parameterized statements and PostgreSQL-native types.
+
+---
+
+## Core System Directives for AI Code Generation
+
+### 1. Data Immutability vs. Human Corrections
+* `job_source.raw_transcription` and `source.raw_transcription` represent original, point-in-time machine outputs and are **immutable**.
+* Human corrections occur on `source.revised_text`.
+* When fetching text for the UI, always display `COALESCE(source.revised_text, source.raw_transcription)`.
+
+### 2. Async Execution & Batching Rules
+* A `job` represents an overarching execution run for a folder/group of images belonging to a single `document`.
+* Images are submitted to AI APIs **one at a time in rapid succession** using `asyncio` worker pools.
+* Each single-image API call populates a row in `job_source` with its own `status`, `raw_transcription`, `ai_metadata`, and `raw_api_response`.
+* If 9 of 10 pages succeed and 1 fails, `job_source.status` for the failed image becomes `'failed'`, while `job.status` becomes `'partial_success'`. Do not mark the entire batch as failed if partial results exist.
+
+### 3. Entity Relationships
+* **Authors/Recipients:** A `document` can have multiple authors and recipients. Do NOT put direct `author_id` foreign keys on `document`. Query authors/recipients via `document_person` where `role = 'author'` or `role = 'recipient'`.
+* **Page Ordering:** Multi-page documents must always be queried using `ORDER BY page_number ASC`.
+
+### 4. Database Mutations
+* Always use parameterized SQL queries (`$1`, `$2`) to prevent SQL injection.
+* Store datetimes using UTC ISO 8601 strings or native PostgreSQL `TIMESTAMPTZ`.

docs/ver2/V2 Python Pydantic Models.md

@@ -0,0 +1,416 @@
+# SQLModel Table Models
+
+These models implement the canonical [Version 2 database schema](../schema_v2.md). Each schema entity is represented by exactly one `SQLModel` table class. Because `SQLModel` is built on Pydantic and SQLAlchemy, these classes provide application validation and PostgreSQL mappings without parallel row and create models.
+
+Database-generated UUIDs and timestamps are `None` until PostgreSQL supplies their values during insert. The database columns remain non-nullable. `Person.metadata_` maps to the `metadata` column because `metadata` is reserved by SQLAlchemy's declarative API.
+
+```python
+from datetime import date
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID
+
+from pydantic import JsonValue
+from sqlalchemy import Column
+from sqlalchemy import Date
+from sqlalchemy import DateTime
+from sqlalchemy import ForeignKey
+from sqlalchemy import Index
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import Text
+from sqlalchemy import UniqueConstraint
+from sqlalchemy import text
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import UUID as PostgreSQLUUID
+from sqlmodel import Field
+from sqlmodel import Relationship
+from sqlmodel import SQLModel
+
+
+class PersonRole(StrEnum):
+    AUTHOR = "author"
+    RECIPIENT = "recipient"
+
+
+class JobStatus(StrEnum):
+    QUEUED = "queued"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    PARTIAL_SUCCESS = "partial_success"
+    FAILED = "failed"
+
+
+class JobSourceStatus(StrEnum):
+    PENDING = "pending"
+    TRANSCRIBED = "transcribed"
+    FAILED = "failed"
+
+
+class Person(SQLModel, table=True):
+    __tablename__ = "person"
+    __table_args__ = (Index("idx_person_full_name", "full_name"),)
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    full_name: str = Field(sa_column=Column(Text, nullable=False))
+    display_name: str | None = Field(default=None, sa_column=Column(Text))
+    maiden_name: str | None = Field(default=None, sa_column=Column(Text))
+    birth_date: date | None = Field(default=None, sa_column=Column(Date))
+    birth_date_raw: str | None = Field(default=None, sa_column=Column(Text))
+    birth_place: str | None = Field(default=None, sa_column=Column(Text))
+    death_date: date | None = Field(default=None, sa_column=Column(Date))
+    death_date_raw: str | None = Field(default=None, sa_column=Column(Text))
+    death_place: str | None = Field(default=None, sa_column=Column(Text))
+    biography: str | None = Field(default=None, sa_column=Column(Text))
+    portrait_path: str | None = Field(default=None, sa_column=Column(Text))
+    metadata_: JsonValue | None = Field(
+        default_factory=dict,
+        sa_column=Column(
+            "metadata",
+            JSONB,
+            server_default=text("'{}'::jsonb"),
+        ),
+    )
+    created_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+    updated_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+
+    document_people: list["DocumentPerson"] = Relationship(
+        back_populates="person",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+
+
+class Document(SQLModel, table=True):
+    __tablename__ = "document"
+    __table_args__ = (Index("idx_document_date", "document_date"),)
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    name: str = Field(sa_column=Column(Text, nullable=False))
+    document_type: str | None = Field(default=None, sa_column=Column(Text))
+    document_date: date | None = Field(default=None, sa_column=Column(Date))
+    document_date_raw: str | None = Field(default=None, sa_column=Column(Text))
+    location_created: str | None = Field(default=None, sa_column=Column(Text))
+    notes: str | None = Field(default=None, sa_column=Column(Text))
+    archive_identifier: str | None = Field(default=None, sa_column=Column(Text))
+    created_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+    updated_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+
+    document_people: list["DocumentPerson"] = Relationship(
+        back_populates="document",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+    jobs: list["Job"] = Relationship(
+        back_populates="document",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+    sources: list["Source"] = Relationship(
+        back_populates="document",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+
+
+class DocumentPerson(SQLModel, table=True):
+    __tablename__ = "document_person"
+    __table_args__ = (
+        UniqueConstraint(
+            "document_id",
+            "person_id",
+            "role",
+            name="unique_document_person_role",
+        ),
+        Index("idx_document_person_doc", "document_id"),
+        Index("idx_document_person_per", "person_id"),
+    )
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    document_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("document.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    person_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("person.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    role: PersonRole = Field(sa_column=Column(String(20), nullable=False))
+    created_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+
+    document: Document | None = Relationship(
+        back_populates="document_people",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+    person: Person | None = Relationship(
+        back_populates="document_people",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+
+
+class Job(SQLModel, table=True):
+    __tablename__ = "job"
+    __table_args__ = (Index("idx_job_document", "document_id"),)
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    document_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("document.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    status: JobStatus = Field(
+        default=JobStatus.QUEUED,
+        sa_column=Column(
+            String(50),
+            nullable=False,
+            server_default=text("'queued'"),
+        ),
+    )
+    retry_count: int = Field(
+        default=0,
+        sa_column=Column(
+            Integer,
+            nullable=False,
+            server_default=text("0"),
+        ),
+    )
+    provider: str = Field(sa_column=Column(Text, nullable=False))
+    model: str = Field(sa_column=Column(Text, nullable=False))
+    prompt_name: str | None = Field(default=None, sa_column=Column(Text))
+    date_created: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+    date_updated: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+
+    document: Document | None = Relationship(
+        back_populates="jobs",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+    job_sources: list["JobSource"] = Relationship(
+        back_populates="job",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+
+
+class Source(SQLModel, table=True):
+    __tablename__ = "source"
+    __table_args__ = (
+        Index("idx_source_document", "document_id"),
+        Index("idx_source_page_order", "document_id", "page_number"),
+    )
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    document_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("document.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    page_number: int = Field(
+        default=1,
+        sa_column=Column(
+            Integer,
+            nullable=False,
+            server_default=text("1"),
+        ),
+    )
+    upload_name: str = Field(sa_column=Column(Text, nullable=False))
+    filename: str = Field(sa_column=Column(Text, nullable=False))
+    file_path: str = Field(sa_column=Column(Text, nullable=False))
+    raw_transcription: str | None = Field(default=None, sa_column=Column(Text))
+    revised_text: str | None = Field(default=None, sa_column=Column(Text))
+    date_uploaded: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+    date_revised: datetime | None = Field(
+        default=None,
+        sa_column=Column(DateTime(timezone=True)),
+    )
+
+    document: Document | None = Relationship(
+        back_populates="sources",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+    job_sources: list["JobSource"] = Relationship(
+        back_populates="source",
+        sa_relationship_kwargs={"lazy": "raise", "passive_deletes": True},
+    )
+
+
+class JobSource(SQLModel, table=True):
+    __tablename__ = "job_source"
+    __table_args__ = (
+        UniqueConstraint("job_id", "source_id", name="unique_job_source"),
+        Index("idx_job_source_job", "job_id"),
+        Index("idx_job_source_source", "source_id"),
+        Index(
+            "idx_job_source_ai_metadata",
+            "ai_metadata",
+            postgresql_using="gin",
+        ),
+    )
+
+    id: UUID | None = Field(
+        default=None,
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            primary_key=True,
+            server_default=text("gen_random_uuid()"),
+        ),
+    )
+    job_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("job.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    source_id: UUID = Field(
+        sa_column=Column(
+            PostgreSQLUUID(as_uuid=True),
+            ForeignKey("source.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+    )
+    status: JobSourceStatus = Field(
+        default=JobSourceStatus.PENDING,
+        sa_column=Column(
+            String(50),
+            nullable=False,
+            server_default=text("'pending'"),
+        ),
+    )
+    raw_transcription: str | None = Field(default=None, sa_column=Column(Text))
+    ai_metadata: JsonValue | None = Field(
+        default=None,
+        sa_column=Column(JSONB),
+    )
+    raw_api_response: JsonValue | None = Field(
+        default=None,
+        sa_column=Column(JSONB),
+    )
+    error_detail: str | None = Field(default=None, sa_column=Column(Text))
+    executed_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=text("now()"),
+        ),
+    )
+
+    job: Job | None = Relationship(
+        back_populates="job_sources",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+    source: Source | None = Relationship(
+        back_populates="job_sources",
+        sa_relationship_kwargs={"lazy": "raise"},
+    )
+```
+
+The enum annotations validate application values while the mapped columns retain the `VARCHAR` types specified by the DDL. PostgreSQL owns generated UUIDs and timestamps through `server_default`; call `session.refresh(instance)` after a flush or commit when those generated values are needed immediately.
+
+`ai_metadata`, `raw_api_response`, and `metadata_` accept any JSON value supported by `JSONB`. Validate provider-specific payload structure before assigning it to these fields, while preserving the complete raw response in `raw_api_response`.
+
+Relationships use `lazy="raise"` to prevent implicit database I/O in async code. Queries must explicitly load relationships they need, for example with `selectinload()`.
+
+The schema's behavioral invariants are enforced outside the table shape where appropriate:
+
+- `PersonRole`, `JobStatus`, and `JobSourceStatus` define the exact values listed by the schema.
+- `unique_document_person_role` enforces role uniqueness for `(document_id, person_id, role)`.
+- Services order document sources by `Source.document_id` and `Source.page_number`.
+- Services derive aggregate `Job.status` from related `JobSource.status` values.
+- Services preserve `JobSource.raw_transcription` and `JobSource.raw_api_response` as point-in-time outputs while updating the active text on `Source`.

src/transcription/__main__.py

@@ -0,0 +1,20 @@
+import uvicorn
+
+from .config import LOGGING_CONFIG
+from .config import get_settings
+
+
+def main() -> None:
+    settings = get_settings()
+    uvicorn.run(
+        "transcription.app:create_app",
+        factory=True,
+        host=settings.host,
+        port=settings.port,
+        log_level=LOGGING_CONFIG.get("root", {}).get("level", "info").lower(),
+        reload=settings.reload,
+    )
+
+
+if __name__ == "__main__":
+    main()

src/transcription/app.py

@@ -16,11 +16,14 @@ from fastapi.staticfiles import StaticFiles
 
 from .api.errors import register_error_handlers
 from .api.health import router as health_router
+from .config import Settings
 from .config import configure_logging
 from .config import get_settings
 from .db import create_all
-from .db import dispose_database_runtime
 from .db import initialize_database_runtime
+from .db.engine import get_database_url
+from .db.engine import resolve_engine
+from .db.session import dispose_session_factory
 from .services import ServiceBundle
 from .services.jobs import JobService
 from .ui import register_pages
@@ -39,7 +42,7 @@ async def _lifespan(app: FastAPI):
     app.state.runtime = initialize_database_runtime(settings=settings)
 
     if settings.should_bootstrap_schema:
-        await create_all(engine=app.state.runtime.engine)
+        await create_all(engine=resolve_engine(settings=settings))
 
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
@@ -47,7 +50,10 @@ async def _lifespan(app: FastAPI):
     await _recover_stale_processing_jobs(app)
 
     async with AsyncExitStack() as stack:
-        stack.push_async_callback(dispose_database_runtime)
+        stack.push_async_callback(
+            dispose_session_factory,
+            database_url=get_database_url(settings),
+        )
         stop_event, worker_notifier = await stack.enter_async_context(
             worker_consumer_lifespan(
                 session_factory=app.state.runtime.session_factory,
@@ -73,14 +79,14 @@ async def _recover_stale_processing_jobs(app: FastAPI) -> None:
         logger.warning("Recovered %s stale processing job(s) at startup", recovered)
 
 
-def create_app() -> FastAPI:
+def create_app(settings: Settings | None = None) -> FastAPI:
     """Create and configure the FastAPI application."""
     app = FastAPI(title="Transcription", lifespan=_lifespan)
-    settings = get_settings()
+    active_settings = settings or get_settings()
-    app.state.settings = settings
+    app.state.settings = active_settings
     app.mount(
         "/uploads",
-        StaticFiles(directory=settings.upload_dir, check_dir=False),
+        StaticFiles(directory=active_settings.upload_dir, check_dir=False),
         name="uploads",
     )
 
@@ -92,6 +98,10 @@ def create_app() -> FastAPI:
     async def ui_redirect() -> RedirectResponse:
         return RedirectResponse(url="/ui/upload", status_code=status.HTTP_307_TEMPORARY_REDIRECT)
 
+    @app.get("/healthz")
+    def health() -> dict[str, str]:
+        return {"status": "ok"}
+
     register_error_handlers(app)
     register_pages(app)
     app.include_router(health_router)

src/transcription/app_state.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.db.runtime import DatabaseRuntime
-from transcription.db.runtime import get_session_factory
+from transcription.db.session import get_session_factory
 from transcription.worker import WorkerNotifier
 from transcription.worker import resolve_worker_notifier
 

src/transcription/config.py

@@ -6,12 +6,16 @@ are resolved by the provider adapters, not here.
 """
 
 import logging.config
-from contextvars import ContextVar
 from enum import StrEnum
+from functools import cache
 from pathlib import Path
+from typing import Annotated
+from typing import Any
 from typing import Literal
 
+from pydantic import BaseModel
 from pydantic import Field
+from pydantic import SecretStr
 from pydantic_settings import BaseSettings
 from pydantic_settings import SettingsConfigDict
 
@@ -22,13 +26,42 @@ class Provider(StrEnum):
     OPENROUTER = "openrouter"
 
 
+class SqliteSettings(BaseModel):
+    driver: Literal["sqlite"] = "sqlite"
+    path: str = "app.db"
+
+
+class PostgresSettings(BaseModel):
+    driver: Literal["postgres"] = "postgres"
+    host: str
+    port: int = 5432
+    database: str
+    user: str
+    password: SecretStr
+
+
+DatabaseSettings = Annotated[
+    SqliteSettings | PostgresSettings,
+    Field(discriminator="driver"),
+]
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",
         extra="ignore",
+        cli_parse_args=True,
+        cli_implicit_flags=True,
+        cli_kebab_case=True,
     )
 
+    # --- NiceGUI Server ---
+    host: str = "0.0.0.0"
+    port: int = 8000
+    log_level: Literal["critical", "error", "warning", "info", "debug", "trace"] = "info"
+    reload: bool = False
+
     # --- AI provider ---
     provider: Provider = Provider.OPENROUTER
     openrouter_api_key: str
@@ -40,8 +73,9 @@ class Settings(BaseSettings):
     environment: Literal["development", "test", "production"] = "development"
 
     # --- persistence ---
+    database: DatabaseSettings = Field(default_factory=SqliteSettings)
     database_url: str = "sqlite:///./transcription.db"
-    bootstrap_schema_on_startup: bool | None = None
+    bootstrap_schema_on_startup: bool = False
     sqlite_check_same_thread: bool = False
 
     # --- filesystem paths ---
@@ -64,18 +98,12 @@ class Settings(BaseSettings):
         return self.environment in {"development", "test"}
 
 
-_settings: ContextVar[Settings | None] = ContextVar("settings", default=None)
+@cache
-
-
 def get_settings(**kwargs) -> Settings:
-    settings = _settings.get()
+    return Settings(**kwargs)
-    if settings is None:
-        settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
-        _settings.set(settings)
-    return settings
 
 
-LOGGING_CONFIG: dict[str, object] = {
+LOGGING_CONFIG: dict[str, Any] = {
     "version": 1,
     "disable_existing_loggers": False,
     "formatters": {
@@ -105,7 +133,10 @@ LOGGING_CONFIG: dict[str, object] = {
 }
 
 
-def configure_logging() -> None:
+def configure_logging(settings: Settings | None = None) -> None:
     """Configure root logging once at startup."""
-    logging.config.dictConfig(LOGGING_CONFIG)
+    cfg = LOGGING_CONFIG.copy()
+    active_settings = settings or get_settings()
+    cfg["loggers"]["transcription"]["level"] = active_settings.log_level.upper()
+    logging.config.dictConfig(cfg)
     logger.debug("Logging configured")

src/transcription/db/__init__.py

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

src/transcription/db/engine.py

@@ -0,0 +1,60 @@
+from functools import cache
+from typing import Any
+
+from sqlalchemy import URL
+from sqlalchemy import StaticPool
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from ..config import PostgresSettings
+from ..config import Settings
+from ..config import SqliteSettings
+from ..config import get_settings
+
+
+def get_database_url(settings: Settings) -> str:
+    match settings.database:
+        case SqliteSettings(path=path):
+            url = URL.create(
+                drivername="sqlite+aiosqlite",
+                database=path,
+            )
+        case PostgresSettings() as database:
+            url = URL.create(
+                drivername="postgresql+asyncpg",
+                host=database.host,
+                port=database.port,
+                database=database.database,
+                username=database.user,
+                password=database.password.get_secret_value(),
+            )
+    return url.render_as_string(hide_password=False)
+
+
+def resolve_engine(settings: Settings | None = None) -> AsyncEngine:
+    active_settings = settings or get_settings()
+    return get_engine(get_database_url(active_settings))
+
+
+@cache
+def get_engine(database_url: str) -> AsyncEngine:
+    kwargs: dict[str, Any] = {"echo": False, "pool_pre_ping": True}
+    if database_url.startswith("sqlite"):
+        kwargs["connect_args"] = {"check_same_thread": False}
+        if ":memory:" in database_url:
+            kwargs["poolclass"] = StaticPool
+
+    return create_async_engine(database_url, **kwargs)
+
+
+async def dispose_engine(database_url: str) -> None:
+    engine = get_engine(database_url)
+    try:
+        await engine.dispose()
+    finally:
+        get_engine.cache_clear()
+
+
+async def refresh_engine(database_url: str) -> AsyncEngine:
+    await dispose_engine(database_url)
+    return get_engine(database_url)

src/transcription/db/operations.py

@@ -10,13 +10,25 @@ from sqlmodel import SQLModel
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from ..models import Job
+from .engine import resolve_engine
-from ..models import JobStatus
+from .models import Job
-from .runtime import get_engine
+from .models import JobStatus
 
 logger = logging.getLogger(__name__)
 
 
+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.db import models as _models  # noqa: F401
+
+    active_engine = engine or resolve_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)
+
+
 async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
     """Get the next queued job, if any."""
     result = await session.exec(
@@ -28,18 +40,6 @@ async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
     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.
 
@@ -68,12 +68,8 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
                     break
             if not has_unique_source:
                 connection.execute(
-                    text(
+                    text("CREATE UNIQUE INDEX IF NOT EXISTS ux_revision_source_id ON revision(source_id)")
-                        "CREATE UNIQUE INDEX IF NOT EXISTS "
-                        "ux_revision_source_id ON revision(source_id)"
-                    )
                 )
                 logger.warning(
-                    "Applied SQLite compatibility schema patch "
+                    "Applied SQLite compatibility schema patch table=revision unique_index=ux_revision_source_id"
-                    "table=revision unique_index=ux_revision_source_id"
                 )

src/transcription/db/runtime.py

@@ -1,18 +1,16 @@
 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
+from .engine import get_database_url
+from .engine import get_engine
+from .session import get_session_factory
 
 logger = logging.getLogger(__name__)
 
@@ -37,33 +35,6 @@ async def dispose_database_runtime() -> None:
     _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()
@@ -71,33 +42,10 @@ def initialize_database_runtime(*, settings: Settings | None = None) -> Database
         return runtime
 
     active_settings = settings or get_settings()
-    engine = _build_engine(active_settings)
+    database_url = get_database_url(active_settings)
-    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    engine = get_engine(database_url)
+    session_factory = get_session_factory(database_url)
     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/db/session.py

@@ -0,0 +1,79 @@
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from functools import cache
+from typing import Annotated
+
+from fastapi import Depends
+from sqlalchemy.ext.asyncio import AsyncSessionTransaction
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ..config import get_settings
+from .engine import dispose_engine
+from .engine import get_database_url
+from .engine import get_engine
+
+type SessionFactory = async_sessionmaker[AsyncSession]
+
+
+@cache
+def get_session_factory(database_url: str) -> SessionFactory:
+    return async_sessionmaker(
+        bind=get_engine(database_url),
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+
+
+def resolve_session_factory(database_url: str | None = None) -> SessionFactory:
+    return get_session_factory(database_url or get_database_url(get_settings()))
+
+
+type SessionFactoryDep = Annotated[SessionFactory, Depends(resolve_session_factory)]
+
+
+async def dispose_session_factory(database_url: str) -> None:
+    get_session_factory.cache_clear()
+    await dispose_engine(database_url)
+
+
+@asynccontextmanager
+async def session_scope(
+    *,
+    database_url: str | None = None,
+    session: AsyncSession | None = None,
+) -> AsyncGenerator[AsyncSession]:
+    if session is not None:
+        yield session
+        return
+
+    session_factory = resolve_session_factory(database_url)
+    async with session_factory() as owned_session:
+        yield owned_session
+
+
+type SessionScopeDep = Annotated[AsyncSession, Depends(session_scope)]
+
+
+@asynccontextmanager
+async def transaction_scope(
+    *,
+    database_url: str | None = None,
+    session: AsyncSessionTransaction | None = None,
+) -> AsyncGenerator[AsyncSessionTransaction]:
+    match session:
+        case AsyncSession() as async_session:
+            if not async_session.in_transaction():
+                raise RuntimeError("A supplied session must have an active transaction")
+            yield async_session
+            return
+        case AsyncSessionTransaction() as async_transaction:
+            yield async_transaction
+            return
+
+    session_factory = resolve_session_factory(database_url)
+    async with session_factory().begin() as owned_session:
+        yield owned_session
+
+
+type TransactionScopeDep = Annotated[AsyncSessionTransaction, Depends(transaction_scope)]

src/transcription/services/base.py

@@ -8,7 +8,8 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..config import Settings
 from ..config import get_settings
-from ..db.runtime import get_session_factory
+from ..db.session import resolve_session_factory
+from ..db.session import session_scope
 
 
 class ServiceBase(ABC):
@@ -24,19 +25,14 @@ class ServiceBase(ABC):
         queue: asyncio.Queue | None = None,
     ):
         self.settings = get_settings()
-        self.session_factory = session_factory or get_session_factory()
+        self.session_factory = session_factory or resolve_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:
+        async with session_scope(session=session) as active_session:
-            # Reuse the provided session if one is passed in
+            yield active_session
-            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,

src/transcription/services/documents.py

@@ -9,9 +9,9 @@ from sqlalchemy.orm import selectinload
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from ..db.models import Document
 from ..errors import AppError
 from ..errors import ErrorCategory
-from ..models import Document
 from .base import ServiceBase
 
 logger = logging.getLogger(__name__)

src/transcription/services/jobs.py

@@ -7,9 +7,9 @@ from sqlalchemy.orm import selectinload
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from ..models import Job
+from ..db.models import Job
-from ..models import JobStatus
+from ..db.models import JobStatus
-from ..models import Source
+from ..db.models import Source
 from .base import ServiceBase
 
 
@@ -170,11 +170,7 @@ class JobService(ServiceBase):
         ``stale_before`` are considered stale and re-queued.
         """
         async with self._session_scope(session) as _session:
-            query = (
+            query = select(Job).where(Job.status == JobStatus.PROCESSING).where(Job.date_updated < stale_before)
-                select(Job)
-                .where(Job.status == JobStatus.PROCESSING)
-                .where(Job.date_updated < stale_before)
-            )
             stale_jobs = (await _session.exec(query)).all()
             if not stale_jobs:
                 return 0

src/transcription/services/store.py

@@ -11,9 +11,9 @@ from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 
-from ..models import Document
+from ..db.models import Document
-from ..models import Job
+from ..db.models import Job
-from ..models import Source
+from ..db.models import Source
 from .documents import UploadJobResult
 
 logger = logging.getLogger(__name__)

src/transcription/services/transcription.py

@@ -18,11 +18,11 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.config import Settings
 from transcription.config import get_settings
+from transcription.db.models import Job
+from transcription.db.models import Revision
+from transcription.db.models import Source
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
-from transcription.models import Job
-from transcription.models import Revision
-from transcription.models import Source
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
 from transcription.providers import ProviderResponseError

src/transcription/services/workflows.py

@@ -5,13 +5,13 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..config import Settings
 from ..config import get_settings
+from ..db.models import Job
+from ..db.models import JobStatus
+from ..db.models import Source
 from ..errors import AppError
 from ..errors import ErrorCategory
 from ..errors import classify_unexpected_error
 from ..errors import format_error_detail
-from ..models import Job
-from ..models import JobStatus
-from ..models import Source
 from ..providers import TranscriptionResult
 from . import ServiceBundle
 from .transcription import DEFAULT_PROMPT_FILE

src/transcription/ui/components/document_panzoom.py

@@ -10,7 +10,7 @@ from uuid import uuid4
 from nicegui import ui
 
 from transcription.config import get_settings
-from transcription.models import Source
+from transcription.db.models import Source
 
 PANGOZOOM_CDN_URL = "https://unpkg.com/@panzoom/[email protected]/dist/panzoom.min.js"
 UPLOADS_URL_PREFIX = "/uploads"

src/transcription/ui/components/job_detail.py

@@ -6,9 +6,9 @@ import logging
 
 from nicegui import ui
 
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import Revision
+from transcription.db.models import Revision
-from transcription.models import Source
+from transcription.db.models import Source
 from transcription.ui.components.document_panzoom import render_document_panzoom
 from transcription.ui.components.transcript import render_original_transcription_card
 from transcription.ui.components.transcript import render_revision_row

src/transcription/ui/components/transcript.py

@@ -9,8 +9,8 @@ from typing import Any
 
 from nicegui import ui
 
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import Revision
+from transcription.db.models import Revision
 
 type RevisionAction = Callable[[Revision], Awaitable[None] | None]
 

src/transcription/ui/pages/jobs_page.py

@@ -4,19 +4,18 @@ from __future__ import annotations
 
 from uuid import UUID
 
-from fastapi import Request
 from nicegui import ui
 
-from transcription.app_state import resolve_session_factory
+from transcription.db.models import Job
-from transcription.models import Job
+from transcription.db.models import JobStatus
-from transcription.models import JobStatus
+from transcription.db.models import Source
-from transcription.models import Source
 from transcription.services.jobs import JobService
 from transcription.services.transcription import TranscriptionService
 from transcription.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.error_presenter import show_error
 from transcription.ui.components.table.jobs import render_jobs_table
 
+from ...db.session import SessionFactoryDep
 from ..components.document_panzoom import render_document_panzoom
 from ..components.table.jobs import JobTableRow
 from ..components.transcript import render_original_transcription_card
@@ -27,8 +26,7 @@ def register_page() -> None:  # noqa: PLR0915
     """Register jobs list and detail routes."""
 
     @ui.page("/jobs")
-    async def jobs_page(request: Request) -> None:
+    async def jobs_page(session_factory: SessionFactoryDep) -> None:
-        session_factory = resolve_session_factory(request.app.state)
         jobs_service = JobService(session_factory=session_factory)
         render_navigation_header(current_path="/jobs")
 
@@ -51,8 +49,7 @@ def register_page() -> None:  # noqa: PLR0915
         await render_table()
 
     @ui.page("/jobs/{job_id}")
-    async def job_detail_page(job_id: str, request: Request) -> None:  # noqa: PLR0915
+    async def job_detail_page(job_id: str, session_factory: SessionFactoryDep) -> None:  # noqa: PLR0915
-        session_factory = resolve_session_factory(request.app.state)
         jobs_service = JobService(session_factory=session_factory)
         transcription_service = TranscriptionService(session_factory=session_factory)
         render_navigation_header(current_path="/jobs")

src/transcription/ui/pages/upload_page.py

@@ -5,8 +5,7 @@ from __future__ import annotations
 from fastapi import Request
 from nicegui import ui
 
-from transcription.app_state import resolve_session_factory
+from transcription.db import session_scope
-from transcription.db import get_session
 from transcription.services.store import create_upload_job
 from transcription.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.upload import render_upload_widget
@@ -19,10 +18,9 @@ def register_page() -> None:
     @ui.page("/upload", title="Upload Document")
     def upload_page(request: Request) -> None:
         render_navigation_header(current_path="/upload")
-        session_factory = resolve_session_factory(request.app.state)
 
         async def submit_upload(filename: str, file_bytes: bytes):
-            async with get_session(session_factory=session_factory) as session:
+            async with session_scope() as session:
                 return await create_upload_job(
                     filename=filename,
                     file_bytes=file_bytes,

src/transcription/worker.py

@@ -14,7 +14,7 @@ from uuid import UUID
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from transcription.db import get_session
+from transcription.db import session_scope
 from transcription.errors import AppError
 from transcription.errors import classify_unexpected_error
 
@@ -178,7 +178,7 @@ async def process_next_queued_job(
         )
 
     if session is None:
-        async with get_session(session_factory=session_factory) as local_session:
+        async with session_scope(session_factory=session_factory) as local_session:
             return await process_next_queued_job_workflow(services=services, session=local_session)
 
     return await process_next_queued_job_workflow(services=services, session=session)

tests/conftest.py

@@ -13,11 +13,12 @@ from sqlmodel.pool import StaticPool
 
 from transcription.config import Settings
 from transcription.config import get_settings
+from transcription.db.engine import get_database_url
+from transcription.db.engine import get_engine
 from transcription.db.operations import create_all
-from transcription.db.runtime import dispose_database_runtime
+from transcription.db.session import dispose_session_factory
-from transcription.db.runtime import get_engine
+from transcription.db.session import get_session_factory
-from transcription.db.runtime import get_session
+from transcription.db.session import session_scope
-from transcription.db.runtime import get_session_factory
 from transcription.services.documents import DocumentService
 from transcription.services.jobs import JobService
 
@@ -39,23 +40,26 @@ def session():
 async def default_settings():
     """Provide default settings for tests."""
     settings = get_settings(database_url="sqlite:///:memory:")
-    await create_all(engine=get_engine(settings=settings))
+    db_url = get_database_url(settings)
+    await create_all(engine=get_engine(database_url=db_url))
     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:
+    db_url = get_database_url(default_settings)
+    async with session_scope(database_url=db_url) as async_session:
         yield async_session
 
-    await dispose_database_runtime()
+    await dispose_session_factory(db_url)
 
 
 @pytest.fixture
 def default_session_factory(default_settings: Settings):
     """Provide a base fixture for tests that require database access."""
-    session_factory = get_session_factory(settings=default_settings)
+    db_url = get_database_url(default_settings)
+    session_factory = get_session_factory(database_url=db_url)
     return session_factory
 
 

tests/integration/test_pipeline_flow.py

@@ -5,8 +5,8 @@ from pathlib import Path
 import pytest
 
 from transcription.config import Settings
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
 from transcription.providers.base import TranscriptionResult
 from transcription.services.store import create_upload_job
 from transcription.worker import process_next_queued_job

tests/services/test_job_service.py

@@ -2,10 +2,10 @@ from uuid import uuid4
 
 import pytest
 
-from transcription.models import Document
+from transcription.db.models import Document
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
-from transcription.models import Source
+from transcription.db.models import Source
 from transcription.services.documents import DocumentService
 from transcription.services.jobs import JobService
 

tests/services/test_transcription_service.py

@@ -4,10 +4,10 @@ from uuid import uuid4
 
 import pytest
 
-from transcription.models import Document
+from transcription.db.models import Document
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
-from transcription.models import Source
+from transcription.db.models import Source
 from transcription.services.documents import DocumentService
 from transcription.services.jobs import JobService
 from transcription.services.transcription import TranscriptionService

tests/services/test_workflows_reliability.py

@@ -6,10 +6,10 @@ from uuid import uuid4
 import pytest
 
 from transcription.config import Settings
-from transcription.models import Document
+from transcription.db.models import Document
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
-from transcription.models import Source
+from transcription.db.models import Source
 from transcription.services import ServiceBundle
 from transcription.services.workflows import process_queued_job
 

tests/test_db.py

@@ -4,17 +4,18 @@ import pytest
 from sqlalchemy import inspect
 
 from transcription.config import Settings
+from transcription.config import SqliteSettings
 from transcription.db import create_all
 from transcription.db import dispose_database_runtime
-from transcription.db import get_session
 from transcription.db import initialize_database_runtime
+from transcription.db import session_scope
 
 
 @pytest.mark.asyncio
 async def test_create_all_creates_expected_tables(tmp_path):
     settings = Settings(
         openrouter_api_key="test-key",
-        database_url=f"sqlite:///{tmp_path / 'schema.db'}",
+        database=SqliteSettings(path=str(tmp_path / "schema.db")),
         environment="test",
     )
     runtime = initialize_database_runtime(settings=settings)
@@ -36,13 +37,13 @@ async def test_create_all_creates_expected_tables(tmp_path):
 async def test_get_session_yields_async_session(tmp_path):
     settings = Settings(
         openrouter_api_key="test-key",
-        database_url=f"sqlite:///{tmp_path / 'session.db'}",
+        database=SqliteSettings(path=str(tmp_path / "session.db")),
         environment="test",
     )
     initialize_database_runtime(settings=settings)
 
     try:
-        async with get_session(settings=settings) as session:
+        async with session_scope(settings=settings) as session:
             assert session is not None
     finally:
         await dispose_database_runtime()

tests/test_models.py

@@ -5,11 +5,11 @@ from uuid import UUID
 import pytest
 from sqlalchemy.exc import IntegrityError
 
-from transcription.models import Document
+from transcription.db.models import Document
-from transcription.models import Job
+from transcription.db.models import Job
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
-from transcription.models import Revision
+from transcription.db.models import Revision
-from transcription.models import Source
+from transcription.db.models import Source
 
 
 def _make_document(**overrides) -> Document:

tests/ui/conftest.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Callable
+from collections.abc import Generator
 from pathlib import Path
 from uuid import UUID
 
@@ -14,32 +15,31 @@ from sqlmodel import delete
 
 from transcription.app import create_app
 from transcription.config import Settings
-from transcription.config import _settings
+from transcription.config import SqliteSettings
 from transcription.db import create_all
-from transcription.db import get_session
 from transcription.db import initialize_database_runtime
-from transcription.models import Document
+from transcription.db import session_scope
-from transcription.models import Job
+from transcription.db.models import Document
-from transcription.models import JobStatus
+from transcription.db.models import Job
-from transcription.models import Revision
+from transcription.db.models import JobStatus
-from transcription.models import Source
+from transcription.db.models import Revision
+from transcription.db.models import Source
 
 RevisionSeed = str
 
 
 @pytest.fixture(scope="session")
-def app_client(tmp_path_factory: pytest.TempPathFactory) -> tuple[FastAPI, TestClient]:
+def app_client(tmp_path_factory: pytest.TempPathFactory) -> Generator[tuple[FastAPI, TestClient]]:
     """Provide a real application and test client backed by in-memory SQLite."""
     tmp_path = tmp_path_factory.mktemp("ui")
     settings = Settings(
         openrouter_api_key="test-key",
-        database_url="sqlite:///:memory:",
+        database=SqliteSettings(path=":memory:"),
         environment="test",
         bootstrap_schema_on_startup=True,
         upload_dir=tmp_path / "uploads",
         prompt_dir=tmp_path / "prompts",
     )
-    _settings.set(settings)
 
     app = create_app()
     app.state.runtime = initialize_database_runtime(settings=settings)
@@ -54,7 +54,7 @@ def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None:
     app, _ = app_client
 
     async def _clear() -> None:
-        async with get_session(session_factory=app.state.runtime.session_factory) as session:
+        async with session_scope() as session:
             await session.exec(delete(Revision))
             await session.exec(delete(Source))
             await session.exec(delete(Job))
@@ -80,7 +80,7 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
         source_file: Path | None = None,
     ) -> UUID:
         async def _insert() -> UUID:
-            async with get_session(session_factory=app.state.runtime.session_factory) as session:
+            async with session_scope() as session:
                 stored_path = app.state.settings.upload_dir / filename
                 stored_path.parent.mkdir(parents=True, exist_ok=True)
                 source_path = source_file or fixtures_dir / "small_png.png"

tests/ui/test_jobs_page.py

@@ -5,7 +5,7 @@ from uuid import uuid4
 
 import pytest
 
-from transcription.models import JobStatus
+from transcription.db.models import JobStatus
 
 
 @pytest.mark.integration