Merge branch 'main' into diverging

Merge branch 'main' of https://gitea.john-stream.com/bbchops/transcription
ver1 - Step6 Implementation Plan
2026-06-26 18:35:27 -05:00 · 2026-06-26 18:19:14 -05:00 · 2026-06-26 18:19:12 -05:00 · 2026-06-26 18:19:09 -05:00 · 2026-06-26 17:51:02 -05:00 · 2026-06-26 15:25:57 -05:00
62 changed files with 2860 additions and 331 deletions
@@ -34,19 +34,62 @@ Optional settings (defaults shown):
 DATABASE_URL=sqlite:///./transcription.db
 UPLOAD_DIR=./uploads
 PROMPT_DIR=./prompts
+MAX_UPLOAD_BYTES=15728640
+OPERATOR_ACCESS_ENABLED=false
+OPERATOR_USERNAME=operator
+# OPERATOR_PASSWORD=replace_with_secure_value
 ```

+
 ### 3) Run the app

 ```bash
 uv run uvicorn transcription.app:create_app --factory --reload
 ```

-### 4) Open in browser
+### 4) (Optional) Run explicit migrations/checks
+
+Use the migration runner for Step 4 schema safety workflows:
+
+```bash
+uv run python -m transcription.migration_runner --list
+uv run python -m transcription.migration_runner --apply
+uv run python -m transcription.migration_runner --check
+```
+
+### 5) Open in browser
+

 - GUI: [http://[IP_ADDRESS]:8000/ui](http://[IP_ADDRESS]:8000/ui)
 - Health check: [http://[IP_ADDRESS]:8000/healthz](http://[IP_ADDRESS]:8000/healthz)

+### Schema safety settings
+
+Optional environment settings (defaults shown):
+
+```env
+MIGRATION_AUTO_APPLY_ON_STARTUP=false
+VALIDATE_SCHEMA_ON_STARTUP=true
+```
+
+### Step 5 security settings
+
+Use this baseline for trusted private-network operation:
+
+```env
+OPERATOR_ACCESS_ENABLED=true
+OPERATOR_USERNAME=operator
+OPERATOR_PASSWORD=replace_with_strong_local_secret
+MAX_UPLOAD_BYTES=15728640
+```
+
+Notes:
+- `/healthz` remains unauthenticated for operational checks.
+- `/ui` and `/api` require HTTP Basic credentials when operator access is enabled.
+- Keep `OPERATOR_PASSWORD` in environment variables only (never commit secrets).
+
+
+
 ## How to navigate the GUI

 - **Upload page** (`/ui`)
@@ -6,7 +6,7 @@ This project is a production application for transcribing and preserving histori

 Read [architecture.md](architecture.md) first.

-Then review [ver1/ver1.md](ver1/ver1.md) for completion scope and [ver1/ver1-step1-results.md](ver1/ver1-step1-results.md) for current architecture-consolidation status.
+Then review [ver1/ver1.md](ver1/ver1.md) for completion scope.

 The architecture page is the primary technical reference and defines:

@@ -41,10 +41,8 @@ This operating model keeps deployment and maintenance simple while preserving cl

 ## Documentation Map

- Architecture and technical design: [architecture.md](architecture.md)
 - Version 1 implementation plan: [ver1/ver1.md](ver1/ver1.md)
- Version 1 Step 1 plan: [ver1/ver1-step1.md](ver1/ver1-step1.md)
- Version 1 Step 1 results: [ver1/ver1-step1-results.md](ver1/ver1-step1-results.md)
+- Architecture and technical design: [architecture.md](architecture.md)
 - Architecture decision records (ADR index): [adr/README.md](adr/README.md)
 - Runtime and deployment requirements: [requirements.md](requirements.md)
 - Error handling policy and operational guidance: [error_handling.md](error_handling.md)
@@ -0,0 +1,113 @@
+# Ver1 Step 4 Migration and Rollback Runbook
+
+## Purpose
+
+Provide a concise, operator-safe procedure for schema migration execution,
+compatibility validation, and rollback/mitigation for personal-scale deployments.
+
+This runbook supports `docs/ver1/ver1-step4.md` and REQ-10 by keeping normal
+production startup non-mutating unless explicitly configured otherwise.
+
+---
+
+## Preconditions
+
+1. Application version to deploy is known and checked out.
+2. `.env` values are configured for target environment.
+3. Database backup path is prepared.
+4. Application process is stopped before migration on production-like systems.
+
+---
+
+## Commands
+
+Use explicit migration runner operations:
+
+1. List pending migrations:
+   - `uv run python -m transcription.migration_runner --list`
+2. Apply pending migrations:
+   - `uv run python -m transcription.migration_runner --apply`
+3. Validate schema compatibility:
+   - `uv run python -m transcription.migration_runner --check`
+
+Recommended execution order:
+
+1. `--list`
+2. backup database
+3. `--apply`
+4. `--check`
+5. start application
+
+---
+
+## Backup Procedure (SQLite Baseline)
+
+For SQLite deployments, copy the DB file before migration:
+
+- Example DB path default: `./transcription.db`
+- Keep timestamped backup copy in a safe location.
+
+If the file is in active use, stop the app first.
+
+---
+
+## Verification Checklist
+
+After migration apply:
+
+1. `--check` exits successfully.
+2. `schema_migration_history` includes applied revisions.
+3. Application starts successfully.
+4. Health endpoint responds: `/healthz`.
+5. Critical flows smoke-check:
+   - upload
+   - job processing
+   - revision listing/acceptance
+
+---
+
+## Rollback and Mitigation Decision Tree
+
+1. If migration fails before changes commit:
+   - fix issue
+   - re-run apply
+2. If migration partially applied or compatibility check fails:
+   - stop app
+   - restore from backup
+   - investigate and produce forward-fix migration if needed
+3. If app starts but functional invariants fail:
+   - stop app
+   - restore backup
+   - add corrective migration/backfill and rehearse before retry
+
+For this Step 4 baseline, backup restore is the primary rollback mechanism.
+
+---
+
+## Failure Classification Guidance
+
+Classify migration failures using `docs/error_handling.md` categories:
+
+- transient connection issues -> `infrastructure_transient_error`
+- permissions/misconfiguration -> `infrastructure_persistent_error`
+- unexpected migration logic defects -> `internal_unexpected_error`
+
+Record failure details with operation context and timestamp.
+
+---
+
+## Operational Notes
+
+- `migration_auto_apply_on_startup` defaults to `False`.
+- `validate_schema_on_startup` defaults to `True`.
+- Startup schema validation fails fast on incompatibility.
+
+This protects production from accidental schema drift.
+
+---
+
+## Post-Step-4 Follow-Up
+
+If migration complexity grows beyond lightweight revision scripts,
+introduce a dedicated migration framework in a future step while preserving
+this runbook structure and operator-first workflow.
@@ -0,0 +1,153 @@
+# Ver1 Step 4 Results: Data Model and Migration Safety
+
+## Summary
+
+Step 4 implementation status: **complete (baseline scope)**.
+
+This document records completed migration-safety work, validation evidence, and remaining follow-ups for Ver1 Step 4.
+
+Implemented in this step:
+
+1. Added explicit migration framework module with revision history tracking.
+2. Added schema compatibility validation and startup guardrails.
+3. Added migration runner CLI for list/apply/check operations.
+4. Added migration tests and Step 4 validation evidence.
+5. Added Step 4 migration/rollback runbook.
+---
+
+## Implemented Changes
+
+### 1) Schema audit and invariant lock
+
+Implemented read-only compatibility checks in `src/transcription/db.py`:
+
+- `validate_schema_compatibility(...)` verifies required V1 tables:
+  - `document`
+  - `job`
+  - `transcript`
+  - `transcriptrevision`
+- verifies required `job.retry_count` column
+- returns explicit issue identifiers (non-mutating check)
+
+### 2) Migration policy/tooling lock
+
+Added explicit migration revision model in `src/transcription/migrations.py`:
+
+- `MigrationRevision` dataclass
+- ordered `MIGRATIONS` registry
+- migration history table: `schema_migration_history`
+- explicit pending-list and apply operations
+
+### 3) Forward migration implementation
+
+Implemented two baseline forward migrations:
+
+1. `0001_add_retry_count_to_job`
+2. `0002_create_transcriptrevision_table`
+
+Each migration is idempotent and recorded in migration history.
+
+### 4) Rollback and mitigation runbook
+
+Created `docs/ver1/ver1-step4-migration-runbook.md` with:
+
+- preconditions
+- list/apply/check command sequence
+- backup-first procedure
+- verification checklist
+- rollback/mitigation decision tree
+- error classification guidance aligned to `docs/error_handling.md`
+
+### 5) Backfill implementation or explicit no-backfill decision
+
+No backfill required for this baseline Step 4 scope.
+
+Rationale:
+
+- additive migration operations only
+- default values and new-table creation do not require historical row rewrites for current V1 invariants
+- residual advanced backfill scenarios deferred unless future schema evolution introduces incompatible transforms
+---
+
+## Test and Verification Evidence
+
+### Added/Updated Tests
+
+1. `tests/test_migrations.py`
+   - pending migration discovery
+   - migration apply + history recording
+   - idempotent re-apply behavior
+2. `tests/test_db.py`
+   - compatibility-check behavior on fresh schema
+   - table expectation updates for `transcriptrevision`
+3. `tests/test_config.py`
+   - migration safety setting defaults
+4. `tests/test_app.py`
+   - lifespan test compatibility with migration/validation startup hooks
+### Validation Runs
+
+Run and record outcomes:
+
+- `uv run pytest --collect-only -q` -> passed
+- `uv run pytest -m unit -q` -> passed
+- `uv run pytest -m "not external" -q` -> passed
+- `uv run pytest -q` -> passed
+### Migration Rehearsal Evidence
+
+Migration rehearsal details (test-based):
+
+- baseline data set used: in-memory SQLite legacy-shaped schema fixture (`job` table missing Step 4 additions)
+- forward migration result: pending revisions applied successfully (`0001`, `0002`)
+- post-migration verification result: schema checks pass and migration history recorded
+- rollback/mitigation rehearsal result: runbook defined backup-restore primary rollback class for personal-scale SQLite deployment
+---
+
+## Requirement Traceability (Step 4)
+
+| Step 4 Area | REQ Coverage | Status | Evidence |
+| --- | --- | --- | --- |
+| Schema lifecycle and state persistence safety | REQ-3, REQ-4, REQ-11 | met | `src/transcription/migrations.py`, `tests/test_migrations.py`, `tests/test_db.py` |
+| Lifespan/runtime ownership continuity | REQ-7 | met | `src/transcription/app.py` startup checks + existing lifespan ownership model |
+| Explicit non-mutating production startup policy | REQ-10 | met | `migration_auto_apply_on_startup=False` default + explicit runner workflow + startup validation gate |
+| Prompt/data continuity constraints | REQ-12 | met (continued) | no prompt-contract mutation in Step 4 changes |
+---
+
+## Operational Artifacts Produced
+
+- `docs/ver1/ver1-step4.md`
+- `docs/ver1/ver1-step4-migration-runbook.md`
+- `src/transcription/migrations.py`
+- `src/transcription/migration_runner.py`
+- README migration workflow updates
+---
+
+## Risks, Exceptions, and Follow-Ups
+
+1. This lightweight migration system is appropriate for current personal-scale scope but may require a dedicated framework as schema complexity grows.
+2. Rollback remains backup-restore primary; reversible down-migration coverage is intentionally limited in this baseline.
+3. Startup compatibility checks currently fail fast with generic runtime error text and can be further normalized under API/operator error envelopes in later hardening.
+
+Open follow-ups to carry forward:
+
+- Evaluate migration framework escalation criteria in Step 9/10 readiness updates.
+- Add optional richer structured migration logging fields if observability scope expands.
+---
+
+## Step 4 Exit Assessment
+
+- Schema validation against finalized V1 domain: **met**
+- Forward migration path safety and repeatability: **met (baseline scope)**
+- Rollback/mitigation readiness: **met (backup-restore primary path)**
+- Backfill risk closure: **met (no backfill required for current deltas)**
+- Test and regression safety: **met**
+
+Step 4 completion status: **complete (baseline scope)**
+---
+
+## Handoff to Step 5
+
+Once Step 4 is marked complete, Step 5 can proceed with:
+
+- verified migration safety baseline
+- explicit rollback and recovery procedures
+- reduced data-integrity risk entering private-network safety hardening
@@ -0,0 +1,378 @@
+# Step 4 Implementation Plan: Data Model and Migration Safety
+
+## Purpose
+
+Implement **Ver1 Step 4** from `docs/ver1/ver1.md` by making data-model evolution safe, explicit, and repeatable for personal-scale deployment.
+
+Step 4 ensures schema changes are handled through deterministic migration workflows rather than implicit startup mutation, while preserving:
+
+- personal-scale operational simplicity
+- single-operator deployment model
+- lifecycle-owned runtime resource boundaries
+- stable requirement traceability and low rollback risk
+
+Primary governing docs:
+
+- `docs/ver1/ver1.md` (Step 4 objective and sequencing)
+- `docs/architecture.md` (runtime ownership, persistence boundaries, simplicity guardrails)
+- `docs/requirements.md` (REQ-3, REQ-4, REQ-7, REQ-10, REQ-11, REQ-12 emphasis)
+- `docs/error_handling.md` (failure classification and safe error surfacing)
+- `docs/intent.md` (verbatim/transcription/revision domain behavior)
+
+---
+
+## MCP Resources Reviewed and Applied
+
+All currently available resources on `john-stream-mcp` were reviewed. Step 4 applies the following guidance directly:
+
+1. `resource://skills/fastapi-async-sqlalchemy-modernization/document`
+   - explicit engine/session lifecycle ownership
+   - transaction boundary clarity for schema transitions and backfills
+   - phased rollout with rollback-aware checkpoints
+
+2. `resource://skills/pydantic-settings/document`
+   - typed migration/runtime safety settings
+   - explicit source-precedence behavior for operational toggles
+   - fail-fast config semantics for unsafe startup paths
+
+3. `resource://skills/pytesting/document`
+   - deterministic migration verification lanes
+   - strict marker discipline
+   - behavior-first test coverage for migration outcomes
+
+4. `resource://skills/python-logging-dictconfig/document`
+   - startup-centralized logging configuration
+   - structured migration and rollback event traceability
+
+5. `resource://skills/fastapi-uv-docker/document`
+   - deployment and rehearsal discipline
+   - startup/health posture validation during migration windows
+
+6. `resource://skills/python-typing/document`
+   - modern typing hygiene for touched migration/persistence modules
+
+7. `resource://skills/ruff-linting-formating/document`
+   - lint/format consistency for migration scripts and database modules
+
+Planning methodology inputs also applied:
+
+8. `resource://prompts/greenfield-architecture/document`
+   - staged execution with explicit risk and extension handling
+
+9. `resource://prompts/pytest-scaffold/document`
+10. `resource://prompts/pytest-fill-scaffold/document`
+   - test-structure-first and deterministic fill-in sequencing
+
+Reviewed but not directly Step 4 execution-critical:
+
+- skills: `copilot-customization`, `mcp-details`, `nicegui`, `nicegui-ui-customization`, `vscode-configuration`, `zensical-docs`
+- prompts: `authoring`, `mcp-consumer-repo-shim`
+
+---
+
+## Current-State Gap Summary (Step 4 Scope)
+
+Based on Step 1–3 outcomes and current docs/tests:
+
+1. **Bootstrap policy baseline is present**
+   - Environment-aware schema bootstrap policy exists and aligns with REQ-10 intent.
+2. **Functional model expanded in Step 3**
+   - Revision/acceptance features introduce schema evolution requirements that need formal migration safety rehearsal.
+3. **Runbook maturity required**
+   - Step 4 requires explicit migration + rollback procedures and evidence.
+4. **Backfill risk must be evaluated**
+   - New/changed fields and semantics must be checked for historical data reconciliation needs.
+5. **Release-path integration needed**
+   - Step 4 artifacts must feed Step 9 release readiness and Step 10 docs completion.
+
+---
+
+## Scope for Step 4
+
+### In scope
+
+1. Validate final V1 schema against implemented domain behavior (post-Step 3 reality).
+2. Define and implement forward-safe migration path for expected upgrades.
+3. Define and document rollback/mitigation strategy for migration failures.
+4. Implement backfill scripts only if required, with idempotent behavior.
+5. Rehearse migration + rollback locally using representative sample data.
+6. Add Step 4-specific verification tests and operational checks.
+7. Produce operator-facing migration/rollback runbook and Step 4 results evidence.
+
+### Out of scope
+
+- Distributed/externally orchestrated migration systems
+- Major persistence-architecture rewrites beyond V1 scope
+- Non-V1 enhancement migrations unrelated to implemented requirement slices
+
+---
+
+## Target Decisions for Step 4
+
+1. **Production startup remains non-mutating by default**
+   - Preserve REQ-10 posture and avoid implicit schema mutation at normal startup.
+
+2. **Schema changes are explicit operator workflows**
+   - Migrations run as deliberate operational actions, not hidden side effects.
+
+3. **Migration safety beats migration speed**
+   - Additive and reversible-first patterns are preferred where possible.
+
+4. **Rollback policy is explicit per change**
+   - Each migration must declare rollback class:
+     - direct rollback supported
+     - forward-fix required
+     - backup restore required
+
+5. **Backfills are optional and minimal**
+   - Introduce only when required by correctness/invariants, never by convenience.
+
+6. **Migration observability is mandatory**
+   - Structured logs include operation, migration identifier, status, and failure classification.
+
+---
+
+## Detailed Work Breakdown
+
+## Phase A — Schema and Domain Invariant Audit
+
+- [ ] **A1. Build canonical V1 schema inventory**
+  - Enumerate all persisted entities and key fields:
+    - document records
+    - jobs and statuses
+    - transcripts
+    - transcript revisions
+    - failure/provenance fields
+- [ ] **A2. Validate invariants against implemented behavior**
+  - Cross-check Step 3 functionality and current domain expectations:
+    - append-only revision history
+    - accepted revision semantics
+    - canonical transcript synchronization behavior
+- [ ] **A3. Classify required schema deltas**
+  - Categorize deltas:
+    - additive and safe
+    - compatibility-sensitive
+    - potentially destructive (must be staged or deferred)
+
+### Deliverables
+
+- `docs/ver1/ver1-step4-schema-audit.md` (recommended)
+- schema-delta matrix with risk class and owning module
+
+### Exit Criteria
+
+- all required schema changes have explicit rationale and risk classification
+- no ambiguous domain invariant remains
+
+---
+
+## Phase B — Migration Policy and Tooling Lock
+
+- [ ] **B1. Lock migration workflow policy**
+  - Define canonical migration execution path and artifact conventions.
+- [ ] **B2. Define migration authoring checklist**
+  - Include:
+    - preconditions
+    - forward steps
+    - rollback class
+    - post-verification checks
+- [ ] **B3. Align policy with runtime startup safeguards**
+  - Ensure production startup remains explicit/non-mutating by default.
+- [ ] **B4. Define operator invocation standard**
+  - One documented command path for local and production-like workflows.
+
+### Deliverables
+
+- migration policy section (this doc + runbook)
+- migration authoring/review checklist
+
+### Exit Criteria
+
+- one unambiguous migration process exists and is documented
+- startup policy and migration policy are consistent and non-conflicting
+
+---
+
+## Phase C — Forward Migration Implementation
+
+- [ ] **C1. Implement required migration set**
+  - Build migration artifacts for all approved Step 4 deltas.
+- [ ] **C2. Preserve compatibility where needed**
+  - Use staged expand/contract strategy when direct cutover is unsafe.
+- [ ] **C3. Add migration logging checkpoints**
+  - Log start, phase boundaries, completion, and failure details.
+- [ ] **C4. Verify post-migration schema state**
+  - Confirm expected tables/columns/constraints/indexes are present.
+
+### Deliverables
+
+- migration artifacts/scripts for V1 target schema
+- schema verification checklist outputs
+
+### Exit Criteria
+
+- baseline-to-target forward migration executes successfully
+- post-migration checks pass deterministically
+
+---
+
+## Phase D — Rollback and Mitigation Strategy
+
+- [ ] **D1. Define rollback classes per migration**
+  - direct downgrade vs forward-fix vs backup-restore.
+- [ ] **D2. Create rollback decision tree**
+  - trigger conditions, safe stop points, and recovery path.
+- [ ] **D3. Align failure classification with `error_handling.md`**
+  - normalize migration failures into canonical categories:
+    - `infrastructure_transient_error`
+    - `infrastructure_persistent_error`
+    - `internal_unexpected_error` (as needed)
+- [ ] **D4. Rehearse rollback flow**
+  - run at least one migration failure simulation and execute chosen recovery path.
+
+### Deliverables
+
+- rollback/mitigation decision tree
+- rehearsal evidence notes
+
+### Exit Criteria
+
+- operator can execute rollback/mitigation without undocumented steps
+- migration failure paths are diagnosable and classified
+
+---
+
+## Phase E — Backfill Decision and Execution (Conditional)
+
+- [ ] **E1. Determine backfill necessity**
+  - inspect whether existing records violate new invariants.
+- [ ] **E2. If required, implement idempotent backfill**
+  - resumable, batch-safe, and deterministic update semantics.
+- [ ] **E3. Add post-backfill verification**
+  - validate:
+    - revision sequencing integrity
+    - accepted/current transcript consistency
+    - job lifecycle consistency
+- [ ] **E4. If not required, record explicit “no backfill needed” evidence**
+
+### Deliverables
+
+- backfill script(s) and checklist (if applicable)
+- no-backfill rationale artifact (if not applicable)
+
+### Exit Criteria
+
+- required backfills completed and verified OR formally ruled out with evidence
+
+---
+
+## Phase F — Verification and Test Expansion
+
+Apply `pytesting` guidance (deterministic, behavior-first, strict markers).
+
+- [ ] **F1. Migration application tests**
+  - verify forward migration from representative baseline.
+- [ ] **F2. Post-migration schema contract tests**
+  - verify expected schema shape and key constraints.
+- [ ] **F3. Rollback/mitigation tests**
+  - verify chosen rollback class behavior where practical.
+- [ ] **F4. Startup policy regression tests**
+  - confirm production-mode startup does not mutate schema implicitly.
+- [ ] **F5. Backfill behavior tests (if applicable)**
+  - idempotency and invariants after repeated execution.
+
+### 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 4 migration-safety checks pass
+- no REQ-10 regression introduced
+
+---
+
+## Phase G — Runbook and Documentation Closure
+
+- [ ] **G1. Create migration and rollback runbook**
+  - include:
+    - prerequisites
+    - backup step
+    - migration execution
+    - verification
+    - rollback/mitigation
+- [ ] **G2. Update traceability artifacts**
+  - map Step 4 outcomes to REQ IDs and evidence.
+- [ ] **G3. Prepare Step 4 handoff artifacts**
+  - ensure outputs feed Step 9 release readiness and Step 10 docs completion.
+
+### Deliverables
+
+- `docs/ver1/ver1-step4-migration-runbook.md` (recommended)
+- `docs/ver1/ver1-step4-results.md`
+- updated traceability references where needed
+
+### Exit Criteria
+
+- migration operations are executable using docs alone
+- Step 4 evidence is complete and auditable
+
+---
+
+## Recommended Implementation Order
+
+1. Phase A — schema/invariant audit
+2. Phase B — migration policy and tooling lock
+3. Phase C — forward migration implementation
+4. Phase D — rollback/mitigation strategy + rehearsal
+5. Phase E — backfill decision and execution (conditional)
+6. Phase F — test and verification expansion
+7. Phase G — runbook + traceability closure
+
+This sequence minimizes risk by locking policy and scope before irreversible data changes.
+
+---
+
+## Risks and Mitigations
+
+1. **Risk:** Data loss from unsafe schema transitions
+   - **Mitigation:** backup-first gate, staged migration strategies, post-check verification.
+
+2. **Risk:** Startup policy drift reintroduces implicit schema mutation
+   - **Mitigation:** explicit regression tests for production startup behavior (REQ-10 guard).
+
+3. **Risk:** Rollback path is incomplete or untested
+   - **Mitigation:** mandatory rollback class declaration + rehearsal evidence.
+
+4. **Risk:** Backfill scripts cause partial/inconsistent state
+   - **Mitigation:** idempotent design, batching, and invariant-focused verification.
+
+5. **Risk:** Migration failure diagnostics are unclear
+   - **Mitigation:** structured logging + error category mapping per `error_handling.md`.
+
+---
+
+## Step 4 Completion Checklist
+
+- [ ] V1 schema audit completed and approved.
+- [ ] Migration workflow policy is locked and documented.
+- [ ] Required forward migrations are implemented and validated.
+- [ ] Rollback/mitigation decision tree is documented and rehearsed.
+- [ ] Backfill required/not-required decision is evidenced.
+- [ ] Migration-safety test coverage is added and passing.
+- [ ] Startup non-mutation policy remains verified in production mode.
+- [ ] Step 4 runbook and results artifacts are completed.
+
+---
+
+## Handoff to Step 5
+
+Step 4 completion enables Step 5 (Private-Network Safety Baseline) with:
+
+- stable, explicit schema evolution mechanics
+- reduced upgrade risk for single-operator deployments
+- migration/rollback procedures suitable for personal-scale production
+- traceable evidence for release-readiness gates
@@ -0,0 +1,178 @@
+# Ver1 Step 5 Results: Private-Network Safety Baseline
+
+## Summary
+
+Step 5 implementation status: **complete**.
+
+This document records completed private-network safety controls, validation evidence, and residual risks for Ver1 Step 5.
+
+Implemented in this step:
+
+1. Added private-network security assumptions and control matrix (`docs/ver1/ver1-step5-security-assumptions.md`).
+2. Implemented optional single-operator access control for `/ui*` and `/api*` via HTTP Basic auth.
+3. Added upload-size guardrails (`MAX_UPLOAD_BYTES`) and config fail-fast validation for operator credential requirements.
+4. Hardened unexpected-error user-facing messaging to reduce sensitive detail leakage.
+5. Added Step 5 tests for access control, security settings, and upload size boundaries.
+6. Executed dependency/security scans (`pip-audit`, `bandit`) with no critical/high findings.
+
+---
+
+## Implemented Changes
+
+### 1) Security assumptions and threat model
+
+Completed.
+
+- Added `docs/ver1/ver1-step5-security-assumptions.md` defining:
+  - trusted private-network deployment assumptions
+  - single-operator usage model
+  - explicit out-of-scope classes (enterprise IAM, internet-facing zero-trust, multi-tenant controls)
+- Added Step 5 control/ownership matrix and residual-risk notes.
+
+### 2) Single-operator access control baseline
+
+Completed.
+
+- New module: `src/transcription/security.py`
+  - `is_protected_path(...)` protects `/ui*` and `/api*`
+  - `enforce_request_access(...)` enforces optional operator auth
+  - robust Basic auth parsing and safe denial responses via `AccessDeniedError`
+- App middleware added in `src/transcription/app.py`:
+  - enforces auth on protected paths
+  - returns consistent `401` envelope and `WWW-Authenticate: Basic` for denied requests
+- Health endpoint `/healthz` remains intentionally unauthenticated.
+
+### 3) Input validation and safe-output hardening
+
+Completed baseline.
+
+- `src/transcription/services/upload.py`
+  - added size-based validation guard (`max_upload_bytes`)
+  - emits `user_input_error` with actionable guidance on over-limit uploads
+- `src/transcription/errors.py`
+  - `classify_unexpected_error(...)` now returns operation-only message without embedding raw exception text
+  - preserves traceability via existing `error_id` and taxonomy while reducing accidental sensitive leak risk
+
+### 4) Secret handling and configuration safety
+
+Completed baseline.
+
+- `src/transcription/config.py` additions:
+  - `max_upload_bytes` (default `15 * 1024 * 1024`)
+  - `operator_access_enabled` (default `False`)
+  - `operator_username` (default `operator`)
+  - `operator_password` (optional, required when auth enabled)
+- Added settings validator enforcing fail-fast config safety:
+  - raises validation error if `OPERATOR_ACCESS_ENABLED=true` and `OPERATOR_PASSWORD` unset
+- `README.md` updated with Step 5 security env settings and explicit secret-handling guidance.
+
+### 5) Dependency/security scanning baseline
+
+Completed.
+
+- Dependency vulnerability scan:
+  - `uvx pip-audit`
+  - Result: **No known vulnerabilities found**
+- Static security scan:
+  - `uvx bandit -r src/transcription`
+  - Result: **No issues identified** (0 low/medium/high)
+
+---
+
+## Test and Verification Evidence
+
+### Added/Updated Tests
+
+1. `tests/api/test_access_control.py`
+   - unauthorized protected API denied (`401` + challenge)
+   - invalid credentials denied
+   - valid credentials accepted
+   - `/ui` protected when auth enabled
+   - `/healthz` remains unprotected
+2. `tests/services/test_upload.py`
+   - added rejection test for payloads above `MAX_UPLOAD_BYTES`
+3. `tests/test_config.py`
+   - added security defaults assertions
+   - added fail-fast assertion for missing `OPERATOR_PASSWORD` when auth enabled
+4. `tests/test_errors.py`
+   - updated expectations for sanitized unexpected-error message behavior
+5. Updated integration expectations where failure detail should no longer include raw exception text:
+   - `tests/services/test_worker.py`
+   - `tests/integration/test_pipeline_flow.py`
+6. `tests/test_app.py` updated for new middleware wiring.
+
+### Validation Runs
+
+Run and record outcomes:
+
+- `uv run pytest --collect-only -q` -> passed
+- `uv run pytest -m unit -q` -> passed
+- `uv run pytest -m "not external" -q` -> passed
+- `uv run pytest -q` -> passed
+
+### Security Scan Evidence
+
+Record scan commands and outcomes:
+
+- dependency scan command(s): `uvx pip-audit`
+- static/security lint command(s): `uvx bandit -r src/transcription`
+- critical/high findings: none
+- remediation/defer decisions: no remediations required for Step 5 baseline
+
+---
+
+## Requirement Traceability (Step 5)
+
+| Step 5 Area | REQ Coverage | Status | Evidence |
+| --- | --- | --- | --- |
+| Private-network and single-operator safety posture | REQ-9 | met | `docs/ver1/ver1-step5-security-assumptions.md`, README security section |
+| Access control behavior at UI/API boundaries | REQ-5, REQ-7 | met | `src/transcription/security.py`, `src/transcription/app.py`, `tests/api/test_access_control.py` |
+| Input validation and safe user-facing error behavior | REQ-1, REQ-2, REQ-5 | met | `src/transcription/services/upload.py`, `src/transcription/errors.py`, updated tests |
+| Config and startup safety controls | REQ-8, REQ-10 | met | `src/transcription/config.py`, `tests/test_config.py`, `README.md` |
+| Persistence and domain integrity continuity | REQ-11, REQ-12 | met (no regressions) | full test lane pass including integration and worker flows |
+
+---
+
+## Operational Artifacts Produced
+
+- `docs/ver1/ver1-step5.md`
+- `docs/ver1/ver1-step5-results.md`
+- `docs/ver1/ver1-step5-security-assumptions.md`
+- `src/transcription/security.py`
+- `tests/api/test_access_control.py`
+
+---
+
+## Risks, Exceptions, and Follow-Ups
+
+1. Basic auth is intentionally right-sized for trusted private-network use; if deployment posture changes, stronger identity controls are required.
+2. Current model remains single shared operator credential (no per-user audit identity).
+3. No built-in brute-force/rate-limit controls in Step 5 scope; evaluate in future hardening if threat model expands.
+
+Open follow-ups to carry forward:
+
+- Consider stronger auth/session model if system becomes multi-user or internet-accessible.
+- Consider request throttling/rate limiting if threat model changes.
+
+---
+
+## Step 5 Exit Assessment
+
+- Private-network assumptions and controls: **met**
+- Access-control baseline effectiveness: **met**
+- Validation and safe-output safety: **met (baseline)**
+- Secret handling and config safety: **met**
+- Dependency/security risk closure: **met (no critical/high findings)**
+- Test and regression safety: **met**
+
+Step 5 completion status: **complete**
+
+---
+
+## Handoff to Step 6
+
+Once Step 5 is marked complete, Step 6 can proceed with:
+
+- clearer operational security assumptions for logs/runbooks
+- hardened boundary behavior for diagnosis and support
+- reduced risk posture for personal-scale ongoing operations
@@ -0,0 +1,51 @@
+# Ver1 Step 5 Security Assumptions (Private-Network Baseline)
+
+## Operating Model
+
+This system is operated as:
+
+1. single operator
+2. trusted private network
+3. non-public deployment (no direct internet exposure for UI/API)
+
+Out of scope for Step 5:
+
+- enterprise IAM/SSO/RBAC
+- internet-facing zero-trust edge controls
+- multi-tenant user isolation
+
+## Step 5 Controls and Ownership
+
+| Control | Boundary Owner | Verification |
+| --- | --- | --- |
+| Optional operator authentication for `/ui*` and `/api*` routes | `src/transcription/security.py`, `src/transcription/app.py` | `tests/api/test_access_control.py` |
+| Unauthorized contract (`401` + safe envelope + `WWW-Authenticate`) | `src/transcription/api/errors.py` | `tests/api/test_access_control.py` |
+| Upload size guard (`MAX_UPLOAD_BYTES`) | `src/transcription/services/upload.py`, `src/transcription/config.py` | `tests/services/test_upload.py` |
+| Fail-fast auth config when enabled | `src/transcription/config.py` | `tests/test_config.py` |
+| Safe unexpected error messaging (reduced leak surface) | `src/transcription/errors.py` | `tests/test_errors.py`, worker/integration failure tests |
+
+## Access-Control Policy (Step 5)
+
+- Health endpoint (`/healthz`) remains unauthenticated for operability checks.
+- When `OPERATOR_ACCESS_ENABLED=true`, protected paths require HTTP Basic auth:
+  - `/ui`
+  - `/ui/...`
+  - `/api/...`
+- Credentials are runtime-configured:
+  - `OPERATOR_USERNAME` (default `operator`)
+  - `OPERATOR_PASSWORD` (required when access is enabled)
+
+## Secrets Policy
+
+- Secrets must be provided via runtime environment variables.
+- Secrets must not be committed to source control.
+- Secrets must not be logged.
+- Example secret values in docs must always be placeholders.
+
+## Residual Risks (Accepted for Step 5)
+
+1. HTTP Basic credentials are suitable only for trusted private-network deployment.
+2. No per-user identity model (single shared operator credential).
+3. No advanced brute-force/rate-limit controls in Step 5 scope.
+
+These are carried forward for future hardening only if deployment posture changes.
@@ -0,0 +1,459 @@
+# Step 5 Implementation Plan: Private-Network Safety Baseline
+
+## Purpose
+
+Implement **Ver1 Step 5** from `docs/ver1/ver1.md` by applying right-sized security controls for a single-user system running on a trusted private network.
+
+Step 5 focuses on practical risk reduction without introducing unnecessary complexity, while preserving:
+
+- personal-scale operational simplicity
+- single-operator workflow
+- explicit boundary ownership from `docs/architecture.md`
+- safety and diagnostics behavior defined in `docs/error_handling.md`
+
+Primary governing docs:
+
+- `docs/ver1/ver1.md` (Step 5 objective and sequencing)
+- `docs/architecture.md` (deployment model and module boundaries)
+- `docs/error_handling.md` (safe user output and diagnostic boundaries)
+- `docs/requirements.md` (REQ-1, REQ-2, REQ-5, REQ-7, REQ-8, REQ-9, REQ-10, REQ-11, REQ-12)
+- `docs/intent.md` (domain integrity priorities)
+
+---
+
+## MCP Resources Reviewed and Applied
+
+All currently available resources on `john-stream-mcp` were reviewed. Step 5 applies the following guidance directly:
+
+1. `resource://skills/pydantic-settings/document`
+   - typed security-related runtime settings
+   - explicit env/source precedence
+   - fail-fast handling for missing/invalid required values
+
+2. `resource://skills/fastapi-uv-docker/document`
+   - environment and deployment safety defaults
+   - startup/health posture and container hygiene assumptions
+   - local secret handling expectations
+
+3. `resource://skills/pytesting/document`
+   - deterministic security-behavior test lanes
+   - marker discipline and behavior-first assertions
+
+4. `resource://skills/python-logging-dictconfig/document`
+   - centralized logging discipline
+   - avoid leaking sensitive values in logs
+
+5. `resource://skills/nicegui-ui-customization/document`
+   - user-safe failure messaging in UI
+   - resilient interaction behavior and clear error feedback
+
+6. `resource://skills/ruff-linting-formating/document`
+   - keep lint quality baseline stable during safety changes
+
+Planning methodology input:
+
+7. `resource://prompts/greenfield-architecture/document`
+   - explicit tradeoff-oriented staging
+   - scope discipline for minimally sufficient security controls
+
+Reviewed but not directly Step 5 execution-critical:
+
+- skills: `copilot-customization`, `fastapi-async-sqlalchemy-modernization`, `mcp-details`, `nicegui`, `python-typing`, `vscode-configuration`, `zensical-docs`
+- prompts: `authoring`, `mcp-consumer-repo-shim`, `pytest-scaffold`, `pytest-fill-scaffold`
+
+---
+
+## Current-State Gap Summary (Step 5 Scope)
+
+Based on current implementation and prior Step outputs:
+
+1. **Private-network assumptions are implicit, not fully codified**
+   - Need explicit, documented security posture and operator constraints.
+
+2. **Access control for UI/API is minimal or absent**
+   - Step 5 requires basic single-operator gating appropriate for private-network use.
+
+3. **Input validation baseline exists but needs security-oriented audit closure**
+   - Upload and API validation should be verified for abuse-resistant boundaries.
+
+4. **Safe error output baseline exists (Step 2), but needs security confirmation pass**
+   - Must ensure no sensitive internals leak through API/UI error payloads.
+
+5. **Secret handling documentation needs formalization in Step 5 artifacts**
+   - Local workflow should clearly prohibit secrets in repo-tracked files and logs.
+
+6. **Dependency/security scanning is not yet formalized as a recurring gate**
+   - Step 5 requires lightweight scanning and triage of high-risk findings.
+
+---
+
+## Scope for Step 5
+
+### In scope
+
+1. Codify private-network and single-operator security assumptions in docs and config.
+2. Add basic access control for UI/API actions (right-sized for trusted network model).
+3. Audit and harden input-validation boundaries (upload, API params/payloads, operational flags).
+4. Verify safe error surface behavior (UI/API) and prevent sensitive leak paths.
+5. Formalize local secret handling policy and usage examples.
+6. Add lightweight dependency/security scan workflow and triage policy.
+7. Add Step 5 verification tests and results artifact.
+
+### Out of scope
+
+- Internet-facing zero-trust security architecture
+- Enterprise IAM/SSO/role systems
+- Full cryptographic key-management infrastructure
+- Major security product integrations beyond lightweight V1 needs
+
+---
+
+## Target Decisions for Step 5
+
+1. **Threat model is explicitly private-network + single operator**
+   - Security controls are right-sized to this posture and documented as assumptions.
+
+2. **Access control is required, even in private network mode**
+   - Basic gate (single shared operator credential/token) protects UI/API mutation paths.
+
+3. **Validation and output safety are strict defaults**
+   - Reject invalid inputs early; never expose sensitive internals in user-facing outputs.
+
+4. **Secrets are runtime-only**
+   - No secrets committed to source control; no plaintext secret logging.
+
+5. **Security scanning is lightweight but mandatory**
+   - Add recurring dependency/security checks with high-risk triage and closure workflow.
+
+6. **No security control may violate Step 1–4 operational simplicity guardrails**
+   - Preserve deployability and maintainability for personal-scale use.
+
+---
+
+## Detailed Work Breakdown
+
+## Phase A — Security Posture Definition and Gap Lock
+
+- [ ] **A1. Define Step 5 threat model**
+  - trusted private network
+  - single operator
+  - local deployment assumptions
+  - explicit out-of-scope threat classes
+
+- [ ] **A2. Produce security baseline checklist**
+  - access control
+  - validation boundaries
+  - safe error behavior
+  - secret handling
+  - dependency risk checks
+
+- [ ] **A3. Map controls to architecture boundaries**
+  - UI
+  - API
+  - service
+  - config/runtime
+  - operator runbooks
+
+### Deliverables
+
+- `docs/ver1/ver1-step5-security-assumptions.md` (recommended)
+- Step 5 control matrix (control -> owner -> validation method)
+
+### Exit Criteria
+
+- private-network safety posture is explicit and approved
+- each in-scope control has boundary ownership and verification path
+
+---
+
+## Phase B — Basic Single-Operator Access Control
+
+- [ ] **B1. Select access mechanism**
+  - minimal approach suitable for private-network model
+  - explicitly document tradeoffs and operator ergonomics
+
+- [ ] **B2. Protect mutating operations first**
+  - upload/create/accept/export-trigger endpoints
+  - UI actions that trigger persistence changes
+
+- [ ] **B3. Protect read operations as policy requires**
+  - determine read-path gating expectations and apply consistently
+
+- [ ] **B4. Add clear unauthorized behavior contract**
+  - stable API status and safe message
+  - UI feedback with actionable operator guidance
+
+### Deliverables
+
+- access-control policy and implementation notes
+- unauthorized behavior matrix (UI/API)
+
+### Exit Criteria
+
+- unauthorized actions are blocked consistently
+- authorized operator flows remain usable and deterministic
+
+---
+
+## Phase C — Input Validation and Safe Output Hardening
+
+- [ ] **C1. Validation audit for all entry points**
+  - file uploads (type/size/content guards)
+  - route/query/body constraints
+  - service-layer invariants
+
+- [ ] **C2. Normalize validation failures to canonical taxonomy**
+  - `validation_error` vs `user_input_error` consistency
+
+- [ ] **C3. Confirm safe error output policy under security lens**
+  - no stack traces/secrets/internal paths in UI/API default outputs
+  - preserve error reference IDs for traceability
+
+- [ ] **C4. Add abuse-resistant guardrails where practical**
+  - basic request-size and payload-shape constraints
+  - anti-duplication interaction safeguards (where missing)
+
+### Deliverables
+
+- validation-path inventory and hardening checklist
+- safe-output verification notes
+
+### Exit Criteria
+
+- input boundaries are deterministic and tested
+- user-facing error outputs remain safe and actionable
+
+---
+
+## Phase D — Secrets Handling and Configuration Safety
+
+- [ ] **D1. Define secret handling policy**
+  - where secrets are allowed (runtime env only)
+  - where secrets are prohibited (source files, docs examples beyond placeholders)
+
+- [ ] **D2. Enforce settings expectations**
+  - required secret fields fail fast
+  - avoid fallback defaults that silently weaken safety
+
+- [ ] **D3. Add operator documentation for local secret workflow**
+  - how to set environment values safely
+  - how to rotate/update credentials locally
+
+- [ ] **D4. Validate logging does not leak secret values**
+  - startup/config logs
+  - error logs for provider/config failures
+
+### Deliverables
+
+- secret-handling section in runbook/README/docs
+- settings and logging safety verification notes
+
+### Exit Criteria
+
+- no secret leakage paths remain in normal operations
+- operator can configure secrets safely using docs only
+
+---
+
+## Phase E — Dependency and Security Scanning Baseline
+
+- [ ] **E1. Select lightweight scanning commands for V1**
+  - dependency vulnerability scan
+  - optional static security scan if practical
+
+- [ ] **E2. Define triage policy for findings**
+  - severity classification
+  - required closure criteria for Step 5 completion
+
+- [ ] **E3. Run scans and capture evidence**
+  - record command outputs/summaries
+  - remediate or formally defer with risk notes
+
+- [ ] **E4. Add recurring execution guidance**
+  - local pre-release checklist integration
+  - future CI gate handoff for Step 7/9
+
+### Deliverables
+
+- Step 5 scan report artifact (recommended)
+- triage log of resolved/deferred findings
+
+### Exit Criteria
+
+- no unresolved critical vulnerabilities in Step 5 scope
+- high-risk findings are resolved or explicitly risk-accepted with rationale
+
+---
+
+## Phase F — Verification and Test Expansion
+
+Apply `pytesting` guidance (deterministic, behavior-first, strict markers).
+
+- [ ] **F1. Access-control tests**
+  - unauthorized requests are rejected as expected
+  - authorized operator requests succeed
+
+- [ ] **F2. Validation and abuse-boundary tests**
+  - invalid payloads rejected with stable category/status
+  - file-type/size constraints enforced
+
+- [ ] **F3. Safe-output tests**
+  - API/UI error responses avoid sensitive details
+  - error IDs and suggestions remain present
+
+- [ ] **F4. Config/secret safety tests**
+  - required secrets fail fast when missing
+  - no unsafe fallback behavior introduced
+
+### 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
+
+- Step 5 safety behavior is test-covered and passing
+- no regression in core upload/transcribe/review workflows
+
+---
+
+## Phase G — Documentation and Risk Closure
+
+- [ ] **G1. Create Step 5 results artifact**
+  - `docs/ver1/ver1-step5-results.md`
+
+- [ ] **G2. Update operator-facing docs**
+  - security assumptions and local deployment cautions
+  - credential handling and recovery basics
+
+- [ ] **G3. Update traceability and carry-forward notes**
+  - map Step 5 controls to REQ and evidence
+
+### Deliverables
+
+- `docs/ver1/ver1-step5-results.md`
+- updated security assumptions checklist and risk summary
+
+### Exit Criteria
+
+- Step 5 controls and residual risks are fully documented
+- handoff is ready for Step 6 observability and Step 7 quality gates
+
+---
+
+## Recommended Implementation Order
+
+1. Phase A — posture definition and gap lock
+2. Phase B — access control baseline
+3. Phase C — validation/output hardening
+4. Phase D — secrets and config safety
+5. Phase E — dependency/security scan baseline
+6. Phase F — test expansion and verification
+7. Phase G — docs and risk closure
+
+This order reduces risk by locking assumptions first, then applying controls at highest-impact boundaries before final verification and documentation.
+
+---
+
+## Step 5 Execution Checklist (Phase-by-Phase)
+
+Use this checklist to execute Step 5 in implementation order and record progress/evidence.
+
+### Phase A — Security Posture Definition and Gap Lock
+
+- [ ] Publish `docs/ver1/ver1-step5-security-assumptions.md`.
+- [ ] Record explicit in-scope and out-of-scope threat classes.
+- [ ] Produce Step 5 control matrix (control, owner, validation method).
+- [ ] Confirm boundary ownership for each control (UI/API/service/config/docs).
+
+### Phase B — Basic Single-Operator Access Control
+
+- [ ] Choose and document access mechanism (with rationale and tradeoffs).
+- [ ] Implement enforcement for mutating API operations.
+- [ ] Implement corresponding UI-side access behavior for protected actions.
+- [ ] Decide and enforce read-path protection policy.
+- [ ] Add unauthorized API/UI contract tests.
+
+### Phase C — Input Validation and Safe Output Hardening
+
+- [ ] Complete input-validation inventory for upload/API/service boundaries.
+- [ ] Tighten payload/file constraints where gaps are found.
+- [ ] Ensure validation failure categories match `docs/error_handling.md`.
+- [ ] Verify user-facing errors remain safe, actionable, and traceable.
+- [ ] Add regression tests for invalid/boundary inputs.
+
+### Phase D — Secrets Handling and Configuration Safety
+
+- [ ] Document secrets policy (runtime-only, no repo storage).
+- [ ] Verify required secret settings fail fast when missing.
+- [ ] Audit logs for accidental secret leakage risk paths.
+- [ ] Update operator docs for local secret setup/rotation workflow.
+- [ ] Add tests for config safety expectations where practical.
+
+### Phase E — Dependency and Security Scanning Baseline
+
+- [ ] Select scanning commands and record tool versions.
+- [ ] Run baseline scans and capture outputs.
+- [ ] Triage findings by severity and exploitability in private-network context.
+- [ ] Resolve/mitigate critical findings; document accepted residual risk.
+- [ ] Add recurring scan guidance for release workflow handoff.
+
+### Phase F — Verification and Test Expansion
+
+- [ ] Run `uv run pytest --collect-only -q`.
+- [ ] Run `uv run pytest -m unit -q`.
+- [ ] Run `uv run pytest -m "not external" -q`.
+- [ ] Run `uv run pytest -q`.
+- [ ] Confirm no regressions in upload/transcribe/review core flows.
+
+### Phase G — Documentation and Risk Closure
+
+- [ ] Complete `docs/ver1/ver1-step5-results.md` with evidence.
+- [ ] Update docs/README/runbooks with final Step 5 security posture.
+- [ ] Record REQ traceability updates and residual risks.
+- [ ] Confirm Step 5 completion checklist items are all closed.
+
+---
+
+## Risks and Mitigations
+
+1. **Risk:** Over-engineering beyond private-network needs
+   - **Mitigation:** enforce Step 5 scope discipline and threat-model constraints.
+
+2. **Risk:** Access controls disrupt operator usability
+   - **Mitigation:** keep mechanism minimal and test primary workflows thoroughly.
+
+3. **Risk:** Sensitive details leak through errors/logging
+   - **Mitigation:** apply safe-output and log-sanitization checks with tests.
+
+4. **Risk:** Unpatched dependency vulnerabilities remain invisible
+   - **Mitigation:** formalize scan + triage + evidence capture workflow.
+
+5. **Risk:** Secret handling remains ad hoc
+   - **Mitigation:** fail-fast settings + explicit operator documentation + review checks.
+
+---
+
+## Step 5 Completion Checklist
+
+- [ ] Private-network and single-operator security assumptions are documented.
+- [ ] Basic single-operator access control is implemented and verified.
+- [ ] Input-validation boundaries are audited, hardened, and test-covered.
+- [ ] UI/API error output safety is confirmed under security tests.
+- [ ] Secret handling policy and local workflow docs are complete.
+- [ ] Dependency/security scans are run; critical findings are resolved.
+- [ ] Step 5 tests pass across all validation lanes.
+- [ ] `docs/ver1/ver1-step5-results.md` is completed with evidence and residual risks.
+
+---
+
+## Handoff to Step 6
+
+Step 5 completion enables Step 6 (Minimal Observability & Operability) with:
+
+- explicit security assumptions for operator context
+- access and validation controls suitable for private-network operation
+- safer runtime/configuration handling for ongoing operations
+- dependency-risk visibility feeding release-readiness gates
@@ -0,0 +1,206 @@
+## Step 6 Goal (from `docs/ver1/ver1.md`)
+Implement **minimal observability & operability** so a single operator can quickly diagnose and recover from common failures.
+
+---
+
+## 1) Current-State Assessment (what already exists)
+
+### Already in place
+- Central startup logging initialization via `setup_logging()` and `dictConfig` (`src/transcription/config.py`, `src/transcription/app.py`).
+- Error taxonomy and `error_id` envelope contract (`src/transcription/errors.py`) aligned with `docs/error_handling.md`.
+- Error handling for API and worker includes category + error IDs in some paths (`src/transcription/api/errors.py`, `src/transcription/worker.py`).
+- Basic health endpoint `/healthz` (`src/transcription/api/health.py`).
+- UI error display already shows actionable message + error reference (`src/transcription/ui/error_presenter.py`).
+
+### Gaps to close for Step 6
+1. **Structured logging is inconsistent** (many logs are free-form text with embedded key/value; no enforced schema).
+2. **Boundary coverage is incomplete** (UI/service/API/worker don’t all emit consistent operation logs).
+3. `/healthz` is very basic; no lightweight readiness/startup diagnostics endpoint/reporting.
+4. No concise **operator runbook** yet (start/stop, log interpretation, recovery playbooks).
+5. Minimal counters/timings are not yet standardized.
+
+---
+
+## 2) MCP Guidance Incorporated (relevant items)
+
+From `john-stream-mcp`, these are directly applied:
+
+- **`python-logging-dictconfig`**: keep one centralized `dictConfig`, configure once at startup, named loggers in modules.
+- **`fastapi-async-sqlalchemy-modernization`**: include observability + health/readiness checks; explicit lifecycle and deterministic startup/shutdown checks.
+- **`fastapi-uv-docker`**: keep `/healthz`; add practical readiness/ops checks for deployment clarity.
+- **`pytesting`**: deterministic tests, concise structure, validation lanes (`collect-only`, `unit`, `not external`, full).
+- **`pydantic-settings`**: keep typed settings as single source for logging/health behavior flags.
+- **`nicegui` + `nicegui-ui-customization`**: preserve clear, actionable user-facing error feedback and non-blocking UI flows.
+- **`zensical-docs`**: produce focused, navigable operator docs.
+
+(Other MCP resources were reviewed but are not core to Step 6 implementation scope.)
+
+---
+
+## 3) Detailed Implementation Plan for Step 6
+
+## Workstream A — Structured Logging Contract
+
+### A1. Define a canonical log event schema
+Create a project log schema (doc + code-level constants) with required keys:
+- `timestamp` (UTC)
+- `level`
+- `logger`
+- `operation`
+- `event`
+- `error_id` (when error)
+- `category` (when error)
+- `exception_type` (when error)
+- `job_id`, `document_id` (when relevant)
+- optional: `duration_ms`, `retry_count`, `status`
+
+### A2. Standardize log emission helpers
+Add small logging helpers (or adapter utilities) to reduce drift:
+- `log_operation_start(...)`
+- `log_operation_success(...)`
+- `log_operation_error(...)`
+
+Keep this minimal and avoid heavy observability frameworks.
+
+### A3. Update formatter to structured output
+Use `dictConfig` to emit either:
+- JSON lines (preferred for structure), or
+- strict key-value line format with fixed fields.
+
+**Recommendation:** JSON lines to satisfy “structured logging” unambiguously while still simple.
+
+---
+
+## Workstream B — Boundary-by-Boundary Instrumentation
+
+### B1. API boundary (`src/transcription/api/*`)
+- Add request-level operation logs for key routes (`upload.submit`, `jobs.list`, `jobs.get`, etc.).
+- Ensure API exception handler logs always include `error_id`, `category`, `operation`, `exception_type`.
+
+### B2. Service boundary (`src/transcription/services/*`)
+- Add operation logs around:
+  - upload validation/persist,
+  - transcription orchestration,
+  - revision add/accept,
+  - search/export.
+- Add timing (`duration_ms`) for high-value operations only.
+
+### B3. Worker boundary (`src/transcription/worker.py`)
+- Standardize all worker log events to schema.
+- Ensure retry logs include: `retriable`, `retry_count`, `max_retries`, `backoff_seconds`.
+- Ensure terminal failure logs include error contract fields.
+
+### B4. UI boundary (`src/transcription/ui/*`)
+- Keep user-safe UI messages as-is.
+- Add backend/UI logger events for user-triggered failures (operation + error_id + category) so UI-visible errors correlate to server logs.
+
+---
+
+## Workstream C — Health, Readiness, Startup Operability
+
+### C1. Keep `/healthz` lightweight
+- Return “process is running” status quickly.
+
+### C2. Add lightweight `/readyz`
+Include small checks:
+- DB connectivity ping.
+- Worker thread alive check.
+- Optional prompt directory existence check.
+
+Return structured status payload with per-check pass/fail.
+
+### C3. Startup self-check summary log
+At startup, emit one concise ops summary event:
+- environment
+- schema validation result
+- worker started
+- directories checked
+- bootstrap/migration mode flags
+
+---
+
+## Workstream D — Minimal Counters & Timings
+
+Add only high-value diagnostics:
+1. `worker_jobs_processed_total`
+2. `worker_jobs_failed_total`
+3. `worker_retries_total`
+4. `transcription_duration_ms` (per job)
+5. `upload_persist_duration_ms` (per upload path)
+
+Implementation can be log-derived counters (no external metrics backend required).
+
+---
+
+## Workstream E — Operator Runbook
+
+Create concise runbook doc (recommended: `docs/ver1/ver1-step6-operator-runbook.md`) with:
+
+1. **Start/Stop**
+   - local `uv` run mode
+   - docker compose mode (if applicable)
+
+2. **Where logs are**
+   - stdout, docker logs commands, filtering by `error_id` / `operation`.
+
+3. **Common failure patterns → recovery**
+   - provider timeout
+   - auth denied
+   - missing prompt dir
+   - DB unavailable
+   - job stuck/failed with retry exhausted
+
+4. **Recovery procedures**
+   - restart sequence
+   - verify health/readiness
+   - when to requeue/re-upload
+
+5. **Escalation artifacts**
+   - capture timestamp + error_id + operation + job_id/document_id
+
+Also update `README.md` with short links to the runbook.
+
+---
+
+## Workstream F — Verification & Quality Gates
+
+### Tests to add/update
+- `tests/api/test_health.py`
+  - `/healthz` baseline
+  - `/readyz` pass/fail behavior
+- `tests/api/test_error_responses.py` / `tests/api/test_routes.py`
+  - logs include `error_id/category/operation` on failures
+- `tests/services/test_worker.py`
+  - retry/failure log fields + timing presence
+- `tests/ui/*`
+  - ensure UI error correlation path includes operation/ref id behavior
+
+### Validation commands (per MCP pytest guidance)
+- `uv run pytest --collect-only -q`
+- `uv run pytest -m unit -q`
+- `uv run pytest -m "not external" -q`
+- `uv run pytest -q`
+
+---
+
+## 4) Traceability to Governing Docs
+
+- **`docs/ver1/ver1.md` Step 6:** all 5 implementation bullets covered.
+- **`docs/error_handling.md`:** logging contract fields and error taxonomy continuity enforced.
+- **`docs/architecture.md`:** respects modular boundaries, in-process worker model, low-complexity ops.
+- **`docs/requirements.md`:**
+  - REQ-8 (startup logging/config centralization) strengthened,
+  - REQ-5 (status visibility) improved operationally,
+  - REQ-7 lifecycle ownership observability improved.
+- **`docs/intent.md`:** keeps operation simple for personal-scale archival workflow.
+
+---
+
+## 5) Suggested Execution Order (low risk)
+
+1. Logging schema + formatter + helpers  
+2. Worker/API instrumentation (highest value)  
+3. Service/UI instrumentation  
+4. `/readyz` + startup summary check  
+5. Runbook + README links  
+6. Tests + Step 6 results artifact (`docs/ver1/ver1-step6-results.md`)
@@ -4,10 +4,13 @@ from __future__ import annotations

 import logging

-from fastapi import FastAPI, Request
+from fastapi import FastAPI
+from fastapi import Request
 from fastapi.responses import JSONResponse

-from transcription.errors import AppError, ErrorCategory, build_error_envelope
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.errors import build_error_envelope

 logger = logging.getLogger(__name__)

@@ -31,6 +34,12 @@ def _status_for(error: AppError) -> int:
 def register_error_handlers(app: FastAPI) -> None:
    """Register API exception handlers on the app."""

+    @app.exception_handler(AccessDeniedError)
+    async def access_denied_handler(_request: Request, exc: AccessDeniedError) -> JSONResponse:
+        envelope = build_error_envelope(exc)
+        headers = {"WWW-Authenticate": "Basic"} if exc.should_challenge else None
+        return JSONResponse(status_code=401, content=envelope.__dict__, headers=headers)
+
    @app.exception_handler(AppError)
    async def app_error_handler(_request: Request, exc: AppError) -> JSONResponse:
        envelope = build_error_envelope(exc)
@@ -5,17 +5,16 @@ from __future__ import annotations
 from uuid import UUID

 from fastapi import APIRouter
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
+from pydantic import Field

-from transcription.services.library import (
-    accept_revision,
-    add_revision,
-    export_transcripts,
-    get_job_detail,
-    list_jobs,
-    list_revisions,
-    search_accepted_transcripts,
-)
+from transcription.services.library import accept_revision
+from transcription.services.library import add_revision
+from transcription.services.library import export_transcripts
+from transcription.services.library import get_job_detail
+from transcription.services.library import list_jobs
+from transcription.services.library import list_revisions
+from transcription.services.library import search_accepted_transcripts

 router = APIRouter(prefix="/api", tags=["transcription"])

@@ -3,30 +3,31 @@
 from __future__ import annotations

 from contextlib import asynccontextmanager
-from threading import Event, Thread
+from threading import Event
+from threading import Thread

 from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession

-from transcription.api.errors import register_error_handlers
-from transcription.api.health import router as health_router
-from transcription.api.routes import router as transcription_router
-from transcription.config import get_settings, setup_logging
-from transcription.db import (
-    create_all,
-    dispose_database_runtime,
-    initialize_database_runtime,
-    should_bootstrap_schema,
-)
-from transcription.ui import register_pages
-from transcription.worker import run_worker_loop
+from .api.errors import register_error_handlers
+from .api.health import router as health_router
+from .config import configure_logging
+from .config import get_settings
+from .db import cleanup_database
+from .db import create_all
+from .db import initialize_database_runtime
+from .ui import register_pages
+from .worker import run_worker_loop


 def _start_worker(app: FastAPI) -> None:
+    session_factory: async_sessionmaker[AsyncSession] = app.state.db_session_factory
    stop_event = Event()
    worker_thread = Thread(
        target=run_worker_loop,
        kwargs={
-            "engine": app.state.db_runtime.engine,
+            "session_factory": session_factory,
            "stop_event": stop_event,
            "poll_interval_seconds": 1.0,
        },
@@ -49,14 +50,16 @@ def _stop_worker(app: FastAPI) -> None:

@asynccontextmanager
 async def _lifespan(app: FastAPI):
-    setup_logging()
+    configure_logging()

    settings = get_settings()
    app.state.settings = settings
-    app.state.db_runtime = initialize_database_runtime(settings=settings)
+    runtime = initialize_database_runtime(settings=settings)
+    app.state.db_engine = runtime.engine
+    app.state.db_session_factory = runtime.session_factory

-    if should_bootstrap_schema(settings):
-        create_all(engine=app.state.db_runtime.engine)
+    if settings.should_bootstrap_schema:
+        await create_all(engine=runtime.engine)

    settings.upload_dir.mkdir(parents=True, exist_ok=True)
    settings.prompt_dir.mkdir(parents=True, exist_ok=True)
@@ -66,15 +69,27 @@ async def _lifespan(app: FastAPI):
        yield
    finally:
        _stop_worker(app)
-        dispose_database_runtime()
+        await cleanup_database()


 def create_app() -> FastAPI:
    """Create and configure the FastAPI application."""
    app = FastAPI(title="Transcription", lifespan=_lifespan)
+
+    @app.middleware("http")
+    async def operator_access_middleware(request: Request, call_next):
+        settings = get_settings()
+        try:
+            enforce_request_access(request=request, settings=settings)
+        except AccessDeniedError as exc:
+            envelope = build_error_envelope(exc)
+            headers = {"WWW-Authenticate": "Basic"} if exc.should_challenge else None
+            return JSONResponse(status_code=401, content=envelope.__dict__, headers=headers)
+
+        return await call_next(request)
+
    register_error_handlers(app)
    register_pages(app)
    app.include_router(health_router)
    app.include_router(transcription_router)
    return app
-
@@ -5,14 +5,16 @@ once at startup. Provider-specific defaults (model names, base URLs)
 are resolved by the provider adapters, not here.
 """

-import logging
 import logging.config
+from contextvars import ContextVar
 from enum import StrEnum
-from functools import lru_cache
 from pathlib import Path
 from typing import Literal

-from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic_settings import BaseSettings
+from pydantic_settings import SettingsConfigDict
+
+logger = logging.getLogger(__name__)


 class Provider(StrEnum):
@@ -39,15 +41,43 @@ class Settings(BaseSettings):
    # --- persistence ---
    database_url: str = "sqlite:///./transcription.db"
    bootstrap_schema_on_startup: bool | None = None
+    migration_auto_apply_on_startup: bool = False
+    validate_schema_on_startup: bool = True

    # --- filesystem paths ---
    upload_dir: Path = Path("./uploads")
    prompt_dir: Path = Path("./prompts")

+    # --- upload safety ---
+    max_upload_bytes: int = 15 * 1024 * 1024
+
+    # --- single-operator access control ---
+    operator_access_enabled: bool = False
+    operator_username: str = "operator"
+    operator_password: str | None = None
+
    # --- worker reliability ---
    worker_max_retries: int = 0
    worker_retry_backoff_seconds: float = 0.0

+    @property
+    def should_bootstrap_schema(self) -> bool:
+        """Return whether startup should auto-create schema for this environment."""
+        if self.bootstrap_schema_on_startup is not None:
+            return self.bootstrap_schema_on_startup
+        return self.environment in {"development", "test"}
+
+
+_settings: ContextVar[Settings | None] = ContextVar("settings", default=None)
+
+
+def get_settings() -> Settings:
+    settings = _settings.get()
+    if settings is None:
+        settings = Settings()  # pyright: ignore[reportCallIssue]
+        _settings.set(settings)
+    return settings
+

 LOGGING_CONFIG: dict[str, object] = {
    "version": 1,
@@ -69,18 +99,17 @@ LOGGING_CONFIG: dict[str, object] = {
        "level": "INFO",
        "handlers": ["console"],
    },
+    "loggers": {
+        "transcription": {
+            "level": "DEBUG",
+            "handlers": ["console"],
+            "propagate": False,
+        }
+    },
 }


-@lru_cache(maxsize=1)
-def get_settings() -> Settings:
-    """Return the singleton Settings instance.
-
-    Cached so the entire application shares one validated config.
-    """
-    return Settings()
-
-
-def setup_logging() -> None:
+def configure_logging() -> None:
    """Configure root logging once at startup."""
    logging.config.dictConfig(LOGGING_CONFIG)
+    logger.debug("Logging configured")
@@ -4,113 +4,146 @@ V1 moves database resource ownership to explicit runtime initialization so
 startup/shutdown behavior is predictable and lifespan-managed.
 """

+from __future__ import annotations
+
 import contextlib
 import logging
-from collections.abc import Generator
+from collections.abc import AsyncGenerator
 from dataclasses import dataclass

-from sqlalchemy import inspect, text
-from sqlalchemy.engine import Engine
-from sqlmodel import Session, SQLModel, create_engine
+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 transcription.config import Settings, get_settings
+from .config import Settings
+from .config import get_settings

 logger = logging.getLogger(__name__)


@dataclass(frozen=True)
 class DatabaseRuntime:
-    """Process-level database runtime resources."""
+    """Database runtime resources owned by app lifespan."""

-    engine: Engine
+    engine: AsyncEngine
+    session_factory: async_sessionmaker[AsyncSession]


 _runtime: DatabaseRuntime | None = None


-def _build_engine(settings: Settings) -> Engine:
+def _to_async_database_url(database_url: str) -> str:
+    """Normalize configured database URL to an async SQLAlchemy driver URL."""
+    if database_url.startswith("sqlite://") and not database_url.startswith("sqlite+aiosqlite://"):
+        return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1)
+    if database_url.startswith("postgresql://") and not database_url.startswith("postgresql+asyncpg://"):
+        return database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+    return database_url
+
+
+def _build_engine(settings: Settings) -> AsyncEngine:
+    database_url = _to_async_database_url(settings.database_url)
    connect_args: dict[str, object] = {}
-    if settings.database_url.startswith("sqlite"):
+    if database_url.startswith("sqlite"):
        connect_args["check_same_thread"] = False
-    return create_engine(
-        settings.database_url,
+    return create_async_engine(
+        url=database_url,
        echo=False,
+        pool_pre_ping=True,
        connect_args=connect_args,
    )


 def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime:
-    """Initialize and cache the process database runtime once."""
+    """Initialize lifespan-owned async DB resources once per process."""
    global _runtime
-
    if _runtime is not None:
        return _runtime

-    runtime_settings = settings or get_settings()
-    _runtime = DatabaseRuntime(engine=_build_engine(runtime_settings))
+    active_settings = settings or get_settings()
+    engine = _build_engine(active_settings)
+    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    _runtime = DatabaseRuntime(engine=engine, session_factory=session_factory)
+    logger.debug("Initialized async database runtime for database_url=%s", engine.url)
    return _runtime


-def get_database_runtime() -> DatabaseRuntime:
-    """Return initialized database runtime, creating it if needed."""
-    if _runtime is None:
-        return initialize_database_runtime()
-    return _runtime
+def get_engine() -> AsyncEngine:
+    """Return the current async SQLAlchemy engine."""
+    runtime = _runtime or initialize_database_runtime()
+    return runtime.engine


-def dispose_database_runtime() -> None:
-    """Dispose process database runtime resources."""
+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 not None:
-        _runtime.engine.dispose()
+    if _runtime is None:
+        return
+    await _runtime.engine.dispose()
    _runtime = None


-def should_bootstrap_schema(settings: Settings) -> bool:
-    """Return whether startup should auto-create schema for this environment."""
-    if settings.bootstrap_schema_on_startup is not None:
-        return settings.bootstrap_schema_on_startup
-    return settings.environment in {"development", "test"}
-
-
-def create_all(*, engine: Engine | None = None) -> None:
+async def create_all(*, engine: AsyncEngine | None = None) -> None:
    """Create all tables on the selected engine."""
    # Import models so SQLModel metadata is fully registered before bootstrap.
    from transcription import models as _models  # noqa: F401

-    active_engine = engine or get_database_runtime().engine
-    SQLModel.metadata.create_all(active_engine)
-    _ensure_sqlite_compat_columns(active_engine)
+    active_engine = engine or get_engine()
+    async with active_engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+        await connection.run_sync(_ensure_sqlite_compat_columns)
+    logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url)


-def _ensure_sqlite_compat_columns(engine: Engine) -> None:
+def _ensure_sqlite_compat_columns(connection: Connection) -> None:
    """Apply lightweight dev/test SQLite compatibility column patches.

-    This keeps local bootstrap resilient when models evolve but no full
-    migration tooling is in place yet.
+    This performs read-only validation and never mutates schema.
    """
-    if engine.url.get_backend_name() != "sqlite":
+    if connection.engine.url.get_backend_name() != "sqlite":
        return

-    inspector = inspect(engine)
+    inspector = inspect(connection)
    table_names = set(inspector.get_table_names())
-    if "job" not in table_names:
-        return
+
+    required_tables = {"document", "job", "transcript", "transcriptrevision"}
+    missing_tables = sorted(required_tables - table_names)
+    for table_name in missing_tables:
+        issues.append(f"missing_table:{table_name}")

    columns = {column["name"] for column in inspector.get_columns("job")}
    if "retry_count" not in columns:
-        with engine.begin() as connection:
-            connection.execute(
-                text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0")
-            )
-        logger.warning(
-            "Applied SQLite compatibility schema patch table=job column=retry_count default=0"
-        )
+        connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
+        logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")


-@contextlib.contextmanager
-def get_session(*, engine: Engine | None = None) -> Generator[Session]:
+@contextlib.asynccontextmanager
+async def get_session(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+) -> AsyncGenerator[AsyncSession]:
    """Yield a database session and ensure cleanup."""
-    active_engine = engine or get_database_runtime().engine
-    with Session(active_engine) as session:
+    active_session_factory = session_factory or get_session_factory()
+    async with active_session_factory() as session:
        yield session
+
+
+def should_bootstrap_schema(settings: Settings) -> bool:
+    """Compatibility helper for explicit bootstrap checks."""
+    return settings.should_bootstrap_schema
@@ -3,7 +3,8 @@
 from __future__ import annotations

 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import UTC
+from datetime import datetime
 from enum import StrEnum
 from uuid import uuid4

@@ -64,14 +65,15 @@ def build_error_envelope(error: AppError) -> ErrorEnvelope:
        category=error.category.value,
        message=error.message,
        suggestion=error.suggestion,
-        timestamp=datetime.now(timezone.utc).isoformat(),
+        timestamp=datetime.now(UTC).isoformat(),
    )


 def classify_unexpected_error(exc: Exception, *, operation: str) -> AppError:
    """Normalize unknown exceptions into internal_unexpected_error."""
+    _ = exc
    return AppError(
-        f"Unexpected error during {operation}: {exc}",
+        f"Unexpected error during {operation}",
        category=ErrorCategory.INTERNAL_UNEXPECTED,
        suggestion="Retry once. If it persists, review logs and report the error reference id.",
        retriable=False,
@@ -83,4 +85,4 @@ def format_error_detail(error: AppError) -> str:
    return (
        f"[{error.category.value}] {error.message} | "
        f"suggestion={error.suggestion} | error_id={error.error_id}"
-    )
+    )
@@ -0,0 +1,75 @@
+"""CLI entrypoint for explicit schema migration and compatibility checks."""
+
+from __future__ import annotations
+
+import argparse
+
+from transcription.config import get_settings
+from transcription.db import initialize_database_runtime
+from transcription.db import validate_schema_compatibility
+from transcription.migrations import apply_pending_migrations
+from transcription.migrations import list_pending_migrations
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Transcription schema migration runner")
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Apply all pending migrations.",
+    )
+    parser.add_argument(
+        "--list",
+        action="store_true",
+        help="List pending migrations.",
+    )
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="Run schema compatibility check.",
+    )
+    return parser
+
+
+def main() -> int:
+    parser = _build_parser()
+    args = parser.parse_args()
+
+    if not (args.apply or args.list or args.check):
+        parser.error("Specify at least one action: --list, --apply, or --check")
+
+    runtime = initialize_database_runtime(settings=get_settings())
+    engine = runtime.engine
+
+    if args.list:
+        pending = list_pending_migrations(engine=engine)
+        if not pending:
+            print("No pending migrations.")
+        else:
+            print("Pending migrations:")
+            for migration in pending:
+                print(f"- {migration.revision_id}: {migration.description}")
+
+    if args.apply:
+        applied = apply_pending_migrations(engine=engine)
+        if not applied:
+            print("No migrations applied.")
+        else:
+            print("Applied migrations:")
+            for revision_id in applied:
+                print(f"- {revision_id}")
+
+    if args.check:
+        issues = validate_schema_compatibility(engine=engine)
+        if issues:
+            print("Schema compatibility check failed:")
+            for issue in issues:
+                print(f"- {issue}")
+            return 1
+        print("Schema compatibility check passed.")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,132 @@
+"""Lightweight schema migration helpers for V1 Step 4.
+
+This module provides explicit, operator-invoked migration execution for
+personal-scale deployments without introducing heavyweight migration tooling.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
+
+from sqlalchemy import inspect
+from sqlalchemy import text
+from sqlalchemy.engine import Connection
+from sqlalchemy.engine import Engine
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class MigrationRevision:
+    """Represents one ordered schema migration revision."""
+
+    revision_id: str
+    description: str
+    apply: Callable[[Connection], None]
+
+
+def _ensure_history_table(connection: Connection) -> None:
+    """Create migration history table when missing."""
+    connection.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS schema_migration_history (
+                revision_id VARCHAR(64) PRIMARY KEY,
+                description VARCHAR(255) NOT NULL,
+                applied_at VARCHAR(64) NOT NULL
+            )
+            """
+        )
+    )
+
+
+def _get_applied_revisions(connection: Connection) -> set[str]:
+    """Return applied migration revision IDs."""
+    _ensure_history_table(connection)
+    rows = connection.execute(text("SELECT revision_id FROM schema_migration_history")).fetchall()
+    return {row[0] for row in rows}
+
+
+def _record_revision(connection: Connection, revision: MigrationRevision) -> None:
+    """Persist one applied migration revision record."""
+    connection.execute(
+        text(
+            """
+            INSERT INTO schema_migration_history (revision_id, description, applied_at)
+            VALUES (:revision_id, :description, :applied_at)
+            """
+        ),
+        {
+            "revision_id": revision.revision_id,
+            "description": revision.description,
+            "applied_at": datetime.now(UTC).isoformat(),
+        },
+    )
+
+
+def _apply_0001_add_retry_count(connection: Connection) -> None:
+    """Ensure job.retry_count exists for legacy databases."""
+    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" in columns:
+        return
+
+    # Compatible with SQLite and PostgreSQL for this additive integer column.
+    connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
+
+
+def _apply_0002_create_transcriptrevision(connection: Connection) -> None:
+    """Ensure transcriptrevision table exists."""
+    # Import models lazily so metadata is fully populated.
+    from sqlmodel import SQLModel
+
+    from transcription.models import TranscriptRevision  # noqa: F401
+
+    table = SQLModel.metadata.tables["transcriptrevision"]
+    table.create(bind=connection, checkfirst=True)
+
+
+MIGRATIONS: tuple[MigrationRevision, ...] = (
+    MigrationRevision(
+        revision_id="0001_add_retry_count_to_job",
+        description="Add retry_count column to job table with default 0",
+        apply=_apply_0001_add_retry_count,
+    ),
+    MigrationRevision(
+        revision_id="0002_create_transcriptrevision_table",
+        description="Create transcriptrevision table for immutable transcript history",
+        apply=_apply_0002_create_transcriptrevision,
+    ),
+)
+
+
+def list_pending_migrations(*, engine: Engine) -> list[MigrationRevision]:
+    """Return pending migrations ordered by revision."""
+    with engine.begin() as connection:
+        applied = _get_applied_revisions(connection)
+    return [revision for revision in MIGRATIONS if revision.revision_id not in applied]
+
+
+def apply_pending_migrations(*, engine: Engine) -> list[str]:
+    """Apply all pending migrations and return applied revision IDs."""
+    pending = list_pending_migrations(engine=engine)
+    applied_ids: list[str] = []
+
+    for revision in pending:
+        logger.info("Applying migration revision=%s", revision.revision_id)
+        with engine.begin() as connection:
+            _ensure_history_table(connection)
+            revision.apply(connection)
+            _record_revision(connection, revision)
+        applied_ids.append(revision.revision_id)
+        logger.info("Applied migration revision=%s", revision.revision_id)
+
+    return applied_ids
@@ -1,11 +1,15 @@
 """SQLModel domain models for the transcription system."""

-from datetime import datetime, timezone
+from datetime import UTC
+from datetime import datetime
 from enum import StrEnum
 from typing import Optional
-from uuid import UUID, uuid4
+from uuid import UUID
+from uuid import uuid4

-from sqlmodel import Field, Relationship, SQLModel
+from sqlmodel import Field
+from sqlmodel import Relationship
+from sqlmodel import SQLModel


 class JobStatus(StrEnum):
@@ -22,7 +26,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(timezone.utc))
+    uploaded_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

    # --- relationships ---
    jobs: list["Job"] = Relationship(back_populates="document")
@@ -35,8 +39,8 @@ 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(timezone.utc))
-    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.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")
@@ -51,7 +55,7 @@ class Transcript(SQLModel, table=True):
    job_id: UUID = Field(foreign_key="job.id", unique=True)
    text: str | None = None
    error_detail: str | None = None
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

    # --- relationships ---
    job: Job = Relationship(back_populates="transcript")
@@ -66,7 +70,7 @@ class TranscriptRevision(SQLModel, table=True):
    text: str
    source: str = Field(default="worker")
    accepted: bool = Field(default=False)
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

    # --- relationships ---
    job: Job = Relationship(back_populates="revisions")
@@ -1,13 +1,13 @@
 """Provider exports and factory for transcription adapters."""

-from transcription.config import Provider, Settings, get_settings
-from transcription.providers.base import (
-    ProviderAuthError,
-    ProviderError,
-    ProviderResponseError,
-    TranscriptionProvider,
-    TranscriptionResult,
-)
+from transcription.config import Provider
+from transcription.config import Settings
+from transcription.config import get_settings
+from transcription.providers.base import ProviderAuthError
+from transcription.providers.base import ProviderError
+from transcription.providers.base import ProviderResponseError
+from transcription.providers.base import TranscriptionProvider
+from transcription.providers.base import TranscriptionResult
 from transcription.providers.openrouter import OpenRouterTranscriptionProvider


@@ -21,11 +21,11 @@ def get_transcription_provider(*, settings: Settings | None = None) -> Transcrip


 __all__ = [
+    "OpenRouterTranscriptionProvider",
    "ProviderAuthError",
    "ProviderError",
    "ProviderResponseError",
    "TranscriptionProvider",
    "TranscriptionResult",
-    "OpenRouterTranscriptionProvider",
    "get_transcription_provider",
 ]
@@ -9,13 +9,12 @@ from typing import Any

 from openrouter import OpenRouter

-from transcription.config import Settings, get_settings
-from transcription.providers.base import (
-    ProviderAuthError,
-    ProviderError,
-    ProviderResponseError,
-    TranscriptionResult,
-)
+from transcription.config import Settings
+from transcription.config import get_settings
+from transcription.providers.base import ProviderAuthError
+from transcription.providers.base import ProviderError
+from transcription.providers.base import ProviderResponseError
+from transcription.providers.base import TranscriptionResult

 logger = logging.getLogger(__name__)

@@ -55,7 +54,7 @@ class OpenRouterTranscriptionProvider:
                http_referer=request.http_referer,
                x_open_router_title=request.x_open_router_title,
            )
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
            message = str(exc).lower()
            if "401" in message or "auth" in message or "api key" in message:
                raise ProviderAuthError("OpenRouter authentication failed") from exc
@@ -111,10 +110,7 @@ class OpenRouterTranscriptionProvider:
            parts: list[str] = []
            for item in content:
                text_part = None
-                if isinstance(item, dict):
-                    text_part = item.get("text")
-                else:
-                    text_part = self._get_optional_attr(item, "text")
+                text_part = item.get("text") if isinstance(item, dict) else self._get_optional_attr(item, "text")

                if isinstance(text_part, str) and text_part.strip():
                    parts.append(text_part.strip())
@@ -0,0 +1,82 @@
+"""Step 5 single-operator access control helpers."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import secrets
+
+from fastapi import Request
+
+from transcription.config import Settings
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+
+
+class AccessDeniedError(AppError):
+    """Raised when a request is not authorized for operator actions."""
+
+    def __init__(self, message: str, *, suggestion: str, should_challenge: bool = True) -> None:
+        super().__init__(message, category=ErrorCategory.USER_INPUT, suggestion=suggestion)
+        self.should_challenge = should_challenge
+
+
+def is_protected_path(path: str) -> bool:
+    """Return True when a request path requires operator authentication."""
+    return path == "/ui" or path.startswith(("/ui/", "/api"))
+
+
+def enforce_request_access(*, request: Request, settings: Settings) -> None:
+    """Enforce basic operator access control for protected paths."""
+    if not settings.operator_access_enabled or not is_protected_path(request.url.path):
+        return
+
+    if not settings.operator_password:
+        raise AppError(
+            "Operator authentication is enabled but credentials are not configured",
+            category=ErrorCategory.INFRA_PERSISTENT,
+            suggestion="Set OPERATOR_PASSWORD in the runtime environment and restart the app.",
+        )
+
+    authorization = request.headers.get("Authorization")
+    username, password = _parse_basic_authorization_header(authorization)
+
+    valid_username = secrets.compare_digest(username, settings.operator_username)
+    valid_password = secrets.compare_digest(password, settings.operator_password)
+    if not (valid_username and valid_password):
+        raise AccessDeniedError(
+            "Invalid operator credentials",
+            suggestion="Provide valid operator credentials and retry.",
+        )
+
+
+def _parse_basic_authorization_header(value: str | None) -> tuple[str, str]:
+    if not value:
+        raise AccessDeniedError(
+            "Operator authentication required",
+            suggestion="Provide HTTP Basic operator credentials and retry.",
+        )
+
+    scheme, _, token = value.partition(" ")
+    if scheme.lower() != "basic" or not token:
+        raise AccessDeniedError(
+            "Operator authentication required",
+            suggestion="Provide HTTP Basic operator credentials and retry.",
+        )
+
+    try:
+        decoded = base64.b64decode(token, validate=True).decode("utf-8")
+    except (binascii.Error, UnicodeDecodeError) as exc:
+        raise AccessDeniedError(
+            "Invalid authentication header",
+            suggestion="Provide HTTP Basic operator credentials and retry.",
+        ) from exc
+
+    username, sep, password = decoded.partition(":")
+    if not sep or not username:
+        raise AccessDeniedError(
+            "Invalid authentication header",
+            suggestion="Provide HTTP Basic operator credentials and retry.",
+        )
+
+    return username, password
@@ -1,30 +1,25 @@
 """Service layer exports."""

-from transcription.services.transcription import (
-    DEFAULT_PROMPT_FILE,
-    PromptLoadError,
-    TranscriptionError,
-    load_image_payload,
-    load_prompt_text,
-    transcribe_document_image,
-)
-from transcription.services.upload import (
-    SUPPORTED_UPLOAD_EXTENSIONS,
-    UploadError,
-    UploadJobResult,
-    create_upload_job,
-)
+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

 __all__ = [
    "DEFAULT_PROMPT_FILE",
+    "SUPPORTED_UPLOAD_EXTENSIONS",
    "PromptLoadError",
    "TranscriptionError",
-    "load_image_payload",
-    "load_prompt_text",
-    "transcribe_document_image",
-    "SUPPORTED_UPLOAD_EXTENSIONS",
    "UploadError",
    "UploadJobResult",
    "create_upload_job",
+    "load_image_payload",
+    "load_prompt_text",
+    "transcribe_document_image",
 ]
-
@@ -3,14 +3,21 @@
 from __future__ import annotations

 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import UTC
+from datetime import datetime
 from uuid import UUID

-from sqlmodel import Session, select
+from sqlmodel import Session
+from sqlmodel import select

 from transcription.db import get_session
-from transcription.errors import AppError, ErrorCategory
-from transcription.models import Document, Job, JobStatus, Transcript, TranscriptRevision
+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 transcription.models import Transcript
+from transcription.models import TranscriptRevision


 class LibraryError(AppError):
@@ -132,7 +139,7 @@ def add_revision(
    transcript.error_detail = None
    session.add(transcript)

-    job.updated_at = datetime.now(timezone.utc)
+    job.updated_at = datetime.now(UTC)
    if accepted:
        job.status = JobStatus.COMPLETED
    elif job.status == JobStatus.QUEUED:
@@ -174,7 +181,7 @@ def accept_revision(*, revision_id: UUID, session: Session | None = None) -> Tra
    job = session.get(Job, revision.job_id)
    if job is not None:
        job.status = JobStatus.COMPLETED
-        job.updated_at = datetime.now(timezone.utc)
+        job.updated_at = datetime.now(UTC)
        session.add(job)

    session.commit()
@@ -6,16 +6,16 @@ import logging
 import mimetypes
 from pathlib import Path

-from transcription.config import Settings, get_settings
-from transcription.errors import AppError, ErrorCategory
-from transcription.providers import (
-    ProviderAuthError,
-    ProviderError,
-    ProviderResponseError,
-    TranscriptionProvider,
-    TranscriptionResult,
-    get_transcription_provider,
-)
+from transcription.config import Settings
+from transcription.config import get_settings
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.providers import ProviderAuthError
+from transcription.providers import ProviderError
+from transcription.providers import ProviderResponseError
+from transcription.providers import TranscriptionProvider
+from transcription.providers import TranscriptionResult
+from transcription.providers import get_transcription_provider

 logger = logging.getLogger(__name__)

@@ -5,14 +5,19 @@ from __future__ import annotations
 import logging
 from dataclasses import dataclass
 from pathlib import Path
-from uuid import UUID, uuid4
+from uuid import UUID
+from uuid import uuid4

-from sqlmodel import Session
+from sqlmodel.ext.asyncio.session import AsyncSession

-from transcription.config import Settings, get_settings
+from transcription.config import Settings
+from transcription.config import get_settings
 from transcription.db import get_session
-from transcription.errors import AppError, ErrorCategory
-from transcription.models import Document, Job, JobStatus
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import JobStatus

 logger = logging.getLogger(__name__)

@@ -33,16 +38,20 @@ class UploadJobResult:
    original_filename: str


-def create_upload_job(
+async def create_upload_job(
    *,
    filename: str,
    file_bytes: bytes,
-    session: Session | None = None,
+    session: AsyncSession | None = None,
    settings: Settings | None = None,
 ) -> UploadJobResult:
    """Persist an uploaded file and create document/job records."""
    runtime_settings = settings or get_settings()
-    _validate_upload(filename=filename, file_bytes=file_bytes)
+    _validate_upload(
+        filename=filename,
+        file_bytes=file_bytes,
+        max_upload_bytes=runtime_settings.max_upload_bytes,
+    )

    upload_dir = runtime_settings.upload_dir
    upload_dir.mkdir(parents=True, exist_ok=True)
@@ -61,15 +70,19 @@ def create_upload_job(

    try:
        if session is not None:
-            document, job = _create_upload_records(session=session, original_filename=filename, stored_path=stored_path)
+            document, job = await _create_upload_records(
+                session=session,
+                original_filename=filename,
+                stored_path=stored_path,
+            )
        else:
-            with get_session() as local_session:
-                document, job = _create_upload_records(
+            async with get_session() as local_session:
+                document, job = await _create_upload_records(
                    session=local_session,
                    original_filename=filename,
                    stored_path=stored_path,
                )
-    except Exception as exc:  # noqa: BLE001
+    except Exception as exc:
        _best_effort_delete(stored_path)
        raise UploadError(
            "Failed to create upload database records",
@@ -87,7 +100,7 @@ def create_upload_job(
    )


-def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
+def _validate_upload(*, filename: str, file_bytes: bytes, max_upload_bytes: int) -> None:
    if not file_bytes:
        raise UploadError(
            "Upload payload is empty",
@@ -95,6 +108,13 @@ def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
            suggestion="Select a non-empty file and try again.",
        )

+    if len(file_bytes) > max_upload_bytes:
+        raise UploadError(
+            f"Upload exceeds maximum allowed size ({max_upload_bytes} bytes)",
+            category=ErrorCategory.USER_INPUT,
+            suggestion="Upload a smaller file or increase MAX_UPLOAD_BYTES for this deployment.",
+        )
+
    safe_name = Path(filename).name
    if not safe_name:
        raise UploadError(
@@ -117,22 +137,27 @@ def _build_stored_filename(filename: str) -> str:
    return f"{uuid4()}_{safe_name}"


-def _create_upload_records(*, session: Session, original_filename: str, stored_path: Path) -> tuple[Document, Job]:
+async def _create_upload_records(
+    *,
+    session: AsyncSession,
+    original_filename: str,
+    stored_path: Path,
+) -> tuple[Document, Job]:
    document = Document(
        filename=Path(original_filename).name,
        file_path=str(stored_path),
    )
    session.add(document)
-    session.flush()
+    await session.flush()

    job = Job(
        document_id=document.id,
        status=JobStatus.QUEUED,
    )
    session.add(job)
-    session.commit()
-    session.refresh(document)
-    session.refresh(job)
+    await session.commit()
+    await session.refresh(document)
+    await session.refresh(job)
    return document, job


@@ -141,4 +166,4 @@ def _best_effort_delete(path: Path) -> None:
        if path.exists():
            path.unlink()
    except OSError:
-        logger.warning("Failed to clean up upload file after DB error: %s", path)
+        logger.warning("Failed to clean up upload file after DB error: %s", path)
@@ -3,8 +3,8 @@
 from fastapi import FastAPI
 from nicegui import ui

-from transcription.ui.jobs_page import register_page as register_jobs_page
-from transcription.ui.upload_page import register_page as register_upload_page
+from transcription.ui.pages.jobs_page import register_page as register_jobs_page
+from transcription.ui.pages.upload_page import register_page as register_upload_page


 def register_pages(app: FastAPI) -> None:
@@ -12,5 +12,3 @@ def register_pages(app: FastAPI) -> None:
    register_upload_page()
    register_jobs_page()
    ui.run_with(app, mount_path="/ui", show_welcome_message=False)
-
-
@@ -4,7 +4,9 @@ from __future__ import annotations

 from nicegui import ui

-from transcription.errors import AppError, ErrorCategory, classify_unexpected_error
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.errors import classify_unexpected_error


 def to_app_error(exc: Exception, *, operation: str) -> AppError:
@@ -37,4 +39,4 @@ def summarize_error(exc: Exception, *, operation: str) -> str:
    error = to_app_error(exc, operation=operation)
    if error.category == ErrorCategory.INTERNAL_UNEXPECTED:
        return f"Unexpected error (ref: {error.error_id})"
-    return f"{error.message} (ref: {error.error_id})"
+    return f"{error.message} (ref: {error.error_id})"
@@ -0,0 +1,30 @@
+"""Reusable job detail rendering helpers."""
+
+from __future__ import annotations
+
+from nicegui import ui
+
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+
+
+def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None:
+    """Render all sections for the job detail page."""
+    ui.label(f"Job ID: {job.id}")
+    ui.label(f"Status: {job.status.value}")
+    ui.label(f"Created: {job.created_at.isoformat()}")
+    ui.label(f"Updated: {job.updated_at.isoformat()}")
+
+    if document is not None:
+        ui.label(f"Filename: {document.filename}")
+        ui.label(f"File path: {document.file_path}")
+
+    if transcript is None:
+        ui.label("Transcript not available yet.")
+    elif transcript.text:
+        ui.label("Transcript:")
+        ui.markdown(transcript.text)
+    elif transcript.error_detail:
+        ui.label("Failure detail:")
+        ui.label(transcript.error_detail)
@@ -0,0 +1,55 @@
+"""Reusable jobs table rendering helpers."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from uuid import UUID
+
+from nicegui import ui
+
+
+@dataclass(frozen=True)
+class JobTableRow:
+    """Read model consumed by the shared jobs table component."""
+
+    id: UUID
+    status: str
+    created_at: str
+    updated_at: str
+
+
+def _serialize_rows(rows: Sequence[JobTableRow]) -> list[dict[str, str]]:
+    """Convert typed rows into table-compatible dictionaries."""
+    return [
+        {
+            "id": str(row.id),
+            "status": row.status,
+            "created_at": row.created_at,
+            "updated_at": row.updated_at,
+        }
+        for row in rows
+    ]
+
+
+def render_jobs_table(rows: Sequence[JobTableRow]) -> None:
+    """Render jobs table and per-row detail links."""
+    if not rows:
+        ui.label("No jobs yet.")
+        return
+
+    serialized_rows = _serialize_rows(rows)
+    ui.table(
+        columns=[
+            {"name": "id", "label": "Job ID", "field": "id"},
+            {"name": "status", "label": "Status", "field": "status"},
+            {"name": "created_at", "label": "Created", "field": "created_at"},
+            {"name": "updated_at", "label": "Updated", "field": "updated_at"},
+        ],
+        rows=serialized_rows,
+        row_key="id",
+    ).classes("w-full")
+
+    with ui.column().classes("gap-1"):
+        for row in serialized_rows:
+            ui.link(f"Open {row['id']}", f"/jobs/{row['id']}")
@@ -9,15 +9,16 @@ from nicegui import ui
 from sqlmodel import select

 from transcription.db import get_session
-from transcription.models import Document, Job, Transcript
-from transcription.services.library import (
-    accept_revision,
-    add_revision,
-    export_transcripts,
-    list_revisions,
-    search_accepted_transcripts,
-)
-from transcription.ui.error_presenter import show_error, summarize_error
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+from transcription.services.library import accept_revision
+from transcription.services.library import add_revision
+from transcription.services.library import export_transcripts
+from transcription.services.library import list_revisions
+from transcription.services.library import search_accepted_transcripts
+from transcription.ui.error_presenter import show_error
+from transcription.ui.error_presenter import summarize_error


@dataclass(frozen=True)
@@ -0,0 +1,92 @@
+"""Jobs list and detail page registration."""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+from nicegui import ui
+from sqlmodel import desc
+from sqlmodel import select
+
+from transcription.db import get_session
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+from transcription.ui.components.error_presenter import show_error
+from transcription.ui.components.error_presenter import summarize_error
+from transcription.ui.components.job_detail import render_job_detail
+from transcription.ui.components.job_table import JobTableRow
+from transcription.ui.components.job_table import render_jobs_table
+
+
+async def fetch_jobs() -> list[JobTableRow]:
+    """Return jobs for display in most-recent-first order."""
+    async with get_session() as session:
+        jobs = (await session.exec(select(Job).order_by(desc(Job.created_at)))).all()
+        return [
+            JobTableRow(
+                id=job.id,
+                status=job.status.value,
+                created_at=job.created_at.isoformat(),
+                updated_at=job.updated_at.isoformat(),
+            )
+            for job in jobs
+        ]
+
+
+async def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]:
+    """Return job, document, and transcript for detail view."""
+    async with get_session() as session:
+        job = await session.get(Job, job_id)
+        if job is None:
+            return None, None, None
+        document = await session.get(Document, job.document_id)
+        transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job.id))).first()
+        return job, document, transcript
+
+
+def register_page() -> None:
+    """Register jobs list and detail routes."""
+
+    @ui.page("/jobs")
+    async def jobs_page() -> None:
+        ui.label("Transcription Jobs")
+        status = ui.label("Ready")
+
+        @ui.refreshable
+        async def render_table() -> None:
+            jobs = await fetch_jobs()
+            render_jobs_table(jobs)
+
+        async def refresh() -> None:
+            status.text = "Refreshing..."
+            try:
+                await render_table.refresh()
+                status.text = "Refreshed"
+            except Exception as exc:  # noqa: BLE001
+                status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}"
+                show_error(exc, title="Jobs refresh failed", operation="jobs.refresh")
+
+        ui.button("Refresh", on_click=refresh)
+        await render_table()
+        ui.link("Back to upload", "/")
+
+    @ui.page("/jobs/{job_id}")
+    async def job_detail_page(job_id: str) -> None:
+        ui.label("Job Detail")
+        try:
+            parsed_id = UUID(job_id)
+        except ValueError:
+            ui.label("Invalid job id")
+            ui.link("Back to jobs", "/jobs")
+            return
+
+        job, document, transcript = await fetch_job_detail(parsed_id)
+        if job is None:
+            ui.label("Job not found")
+            ui.link("Back to jobs", "/jobs")
+            return
+
+        render_job_detail(job=job, document=document, transcript=transcript)
+
+        ui.link("Back to jobs", "/jobs")
@@ -7,8 +7,11 @@ from dataclasses import dataclass
 from nicegui import ui
 from nicegui.events import UploadEventArguments

-from transcription.services.upload import UploadError, UploadJobResult, create_upload_job
-from transcription.ui.error_presenter import show_error, summarize_error
+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
@@ -24,9 +27,9 @@ def accepted_upload_types() -> str:
    return ".jpg,.jpeg,.png,.tif,.tiff,.pdf"


-def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult:
+async def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult:
    """Create an upload job from incoming file data."""
-    return create_upload_job(filename=filename, file_bytes=file_bytes)
+    return await create_upload_job(filename=filename, file_bytes=file_bytes)


 def register_page() -> None:
@@ -46,7 +49,7 @@ def register_page() -> None:
            status_label.text = "Uploading..."
            try:
                payload = await event.file.read()
-                result = submit_upload(filename=event.file.name, file_bytes=payload)
+                result = await submit_upload(filename=event.file.name, file_bytes=payload)
                state.message = f"Created job {result.job_id}"
                status_label.text = state.message
                ui.notify(state.message, type="positive")
@@ -2,54 +2,62 @@

 from __future__ import annotations

+import asyncio
 import logging
-import time
-from datetime import datetime, timezone
+from datetime import UTC
+from datetime import datetime
 from threading import Event

 from pydantic import ValidationError
-from sqlalchemy.engine import Engine
-from sqlmodel import Session, select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession

-from transcription.config import Settings, get_settings
+from transcription.config import Settings
+from transcription.config import get_settings
 from transcription.db import get_session
-from transcription.errors import AppError, ErrorCategory, classify_unexpected_error, format_error_detail
-from transcription.models import Document, Job, JobStatus, Transcript
-from transcription.services.library import add_revision
+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

 logger = logging.getLogger(__name__)


-def process_next_queued_job(*, session: Session | None = None, engine: Engine | None = None) -> bool:
+async def process_next_queued_job(
+    *,
+    session: AsyncSession | None = None,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+) -> bool:
    """Process the next queued job and persist terminal outcome.

    Returns True when a job was processed, False when no queued job exists.
    """
    if session is None:
-        with get_session(engine=engine) as local_session:
-            return _process_next_queued_job(session=local_session)
-    return _process_next_queued_job(session=session)
+        async with get_session(session_factory=session_factory) as local_session:
+            return await _process_next_queued_job(session=local_session)
+    return await _process_next_queued_job(session=session)


-def _process_next_queued_job(*, session: Session) -> bool:
-    job = session.exec(
-        select(Job)
-        .where(Job.status == JobStatus.QUEUED)
-        .order_by(Job.created_at)
-    ).first()
+async def _process_next_queued_job(*, session: AsyncSession) -> bool:
+    job = (await session.exec(select(Job).where(Job.status == JobStatus.QUEUED).order_by(Job.created_at))).first()

    if job is None:
        return False

    logger.info("Picked queued job operation=worker.pick job_id=%s", job.id)
    job.status = JobStatus.PROCESSING
-    job.updated_at = datetime.now(timezone.utc)
+    job.updated_at = datetime.now(UTC)
    session.add(job)
-    session.commit()
-    session.refresh(job)
+    await session.commit()
+    await session.refresh(job)

-    document = session.get(Document, job.document_id)
+    document = await session.get(Document, job.document_id)
    if document is None:
        error = AppError(
            "Document not found",
@@ -67,17 +75,11 @@ def _process_next_queued_job(*, session: Session) -> bool:

    try:
        result = transcribe_document_image(document.file_path)
-        revision = add_revision(
-            job_id=job.id,
-            text=result.text,
-            source="worker",
-            accepted=False,
-            session=session,
-        )
+        await _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None)
        job.status = JobStatus.TRANSCRIBED
-        job.updated_at = datetime.now(timezone.utc)
+        job.updated_at = datetime.now(UTC)
        session.add(job)
-        session.commit()
+        await session.commit()
        logger.info(
            "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s revision_number=%s",
            job.id,
@@ -85,11 +87,11 @@ def _process_next_queued_job(*, session: Session) -> bool:
            result.provider,
            revision.revision_number,
        )
-    except Exception as exc:  # noqa: BLE001
+    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):
-            _requeue_for_retry(session=session, job=job, error=error, settings=settings)
+            await _requeue_for_retry(session=session, job=job, error=error, settings=settings)
            logger.warning(
                "Job retried operation=worker.process_job job_id=%s document_id=%s retry_count=%s error_id=%s category=%s",
                job.id,
@@ -99,7 +101,7 @@ def _process_next_queued_job(*, session: Session) -> bool:
                error.category.value,
            )
        else:
-            _finalize_failed_job(session=session, job=job, error=error)
+            await _finalize_failed_job(session=session, job=job, error=error)
            logger.exception(
                "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
                job.id,
@@ -111,16 +113,18 @@ def _process_next_queued_job(*, session: Session) -> bool:
    return True


-def _upsert_transcript(*, session: Session, job_id, text: str | None, error_detail: str | None) -> Transcript:
-    transcript = session.exec(select(Transcript).where(Transcript.job_id == job_id)).first()
+async def _upsert_transcript(
+    *, session: AsyncSession, job_id, text: str | None, error_detail: str | None
+) -> Transcript:
+    transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job_id))).first()
    if transcript is None:
        transcript = Transcript(job_id=job_id)

    transcript.text = text
    transcript.error_detail = error_detail
    session.add(transcript)
-    session.commit()
-    session.refresh(transcript)
+    await session.commit()
+    await session.refresh(transcript)
    return transcript


@@ -135,32 +139,53 @@ def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool:
    return error.retriable and job.retry_count < settings.worker_max_retries


-def _requeue_for_retry(*, session: Session, job: Job, error: AppError, settings: Settings) -> None:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None:
+    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
    job.retry_count += 1
    job.status = JobStatus.QUEUED
-    job.updated_at = datetime.now(timezone.utc)
+    job.updated_at = datetime.now(UTC)
    session.add(job)
-    session.commit()
+    await session.commit()
    if settings.worker_retry_backoff_seconds > 0:
-        time.sleep(settings.worker_retry_backoff_seconds)
+        await asyncio.sleep(settings.worker_retry_backoff_seconds)


-def _finalize_failed_job(*, session: Session, job: Job, error: AppError) -> None:
-    _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
+async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None:
+    await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error))
    job.status = JobStatus.FAILED
-    job.updated_at = datetime.now(timezone.utc)
+    job.updated_at = datetime.now(UTC)
    session.add(job)
-    session.commit()
+    await session.commit()


-def run_worker_loop(*, engine: Engine | None = None, stop_event: Event | None = None, poll_interval_seconds: float = 1.0) -> None:
+async def _run_worker_loop_async(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    stop_event: Event | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> None:
    """Run worker polling loop until stop_event is set."""
    while True:
        if stop_event is not None and stop_event.is_set():
            logger.info("Worker stop event received")
            return

-        processed = process_next_queued_job(engine=engine)
+        processed = await process_next_queued_job(session_factory=session_factory)
        if not processed:
-            time.sleep(poll_interval_seconds)
+            await asyncio.sleep(poll_interval_seconds)
+
+
+def run_worker_loop(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None = None,
+    stop_event: Event | None = None,
+    poll_interval_seconds: float = 1.0,
+) -> None:
+    """Synchronous thread entrypoint that runs the async worker loop."""
+    asyncio.run(
+        _run_worker_loop_async(
+            session_factory=session_factory,
+            stop_event=stop_event,
+            poll_interval_seconds=poll_interval_seconds,
+        )
+    )
@@ -0,0 +1,129 @@
+"""Tests for Step 5 operator access control behavior."""
+
+from __future__ import annotations
+
+import base64
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+from fastapi.testclient import TestClient
+
+from transcription.api.errors import register_error_handlers
+from transcription.errors import build_error_envelope
+from transcription.security import AccessDeniedError
+from transcription.security import enforce_request_access
+
+
+def _basic_header(username: str, password: str) -> str:
+    token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
+    return f"Basic {token}"
+
+
+def _build_app(*, settings) -> FastAPI:
+    app = FastAPI()
+    register_error_handlers(app)
+
+    @app.middleware("http")
+    async def operator_access_middleware(request, call_next):
+        try:
+            enforce_request_access(request=request, settings=settings)
+        except AccessDeniedError as exc:
+            envelope = build_error_envelope(exc)
+            headers = {"WWW-Authenticate": "Basic"} if exc.should_challenge else None
+            return JSONResponse(status_code=401, content=envelope.__dict__, headers=headers)
+        return await call_next(request)
+
+    @app.get("/healthz")
+    def healthz():
+        return {"status": "ok"}
+
+    @app.get("/api/jobs")
+    def get_jobs():
+        return [{"id": "demo"}]
+
+    @app.get("/ui")
+    def ui_root():
+        return {"ok": True}
+
+    return app
+
+
+@pytest.mark.integration
+class TestAccessControl:
+    """Verify protected routes enforce operator auth when enabled."""
+
+    def test_protected_api_requires_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/api/jobs")
+
+        assert response.status_code == 401
+        assert response.headers.get("WWW-Authenticate") == "Basic"
+        payload = response.json()
+        assert payload["category"] == "user_input_error"
+        assert payload["suggestion"]
+
+    def test_protected_api_rejects_invalid_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get(
+            "/api/jobs",
+            headers={"Authorization": _basic_header("operator", "wrong")},
+        )
+
+        assert response.status_code == 401
+        payload = response.json()
+        assert payload["message"] == "Invalid operator credentials"
+
+    def test_protected_api_allows_valid_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get(
+            "/api/jobs",
+            headers={"Authorization": _basic_header("operator", "secret")},
+        )
+
+        assert response.status_code == 200
+        assert response.json() == [{"id": "demo"}]
+
+    def test_protected_ui_path_requires_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/ui")
+
+        assert response.status_code == 401
+
+    def test_healthz_is_not_protected(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/healthz")
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "ok"}
@@ -1,11 +1,12 @@
 """Tests for API error response envelope handlers."""

+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
-import pytest

 from transcription.api.errors import register_error_handlers
-from transcription.errors import AppError, ErrorCategory
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory


@pytest.mark.integration
@@ -1,12 +1,13 @@
 """Tests for Step 3 functional API routes."""

-from datetime import datetime, timezone
+from datetime import UTC
+from datetime import datetime
 from types import SimpleNamespace
 from uuid import uuid4

+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
-import pytest

 from transcription.api.errors import register_error_handlers
 from transcription.api.routes import router
@@ -25,7 +26,7 @@ class TestFunctionalRoutes:

    def test_get_jobs_returns_serialized_rows(self, monkeypatch):
        """GET /api/jobs returns normalized job rows."""
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
        job = SimpleNamespace(
            id=uuid4(),
            document_id=uuid4(),
@@ -53,7 +54,7 @@ class TestFunctionalRoutes:
            text="edited text",
            source="user",
            accepted=False,
-            created_at=datetime.now(timezone.utc),
+            created_at=datetime.now(UTC),
        )
        monkeypatch.setattr("transcription.api.routes.add_revision", lambda **_kwargs: revision)

@@ -78,7 +79,7 @@ class TestFunctionalRoutes:
            text="family archive",
            source="user",
            accepted=True,
-            created_at=datetime.now(timezone.utc),
+            created_at=datetime.now(UTC),
        )
        monkeypatch.setattr("transcription.api.routes.search_accepted_transcripts", lambda query: [result])

@@ -103,7 +104,7 @@ class TestFunctionalRoutes:
                "accepted": True,
                "source": "user",
                "text": "exported",
-                "created_at": datetime.now(timezone.utc).isoformat(),
+                "created_at": datetime.now(UTC).isoformat(),
            }
        ]
        monkeypatch.setattr("transcription.api.routes.export_transcripts", lambda accepted_only=True: records)
@@ -32,7 +32,7 @@ enough pictures but we had to take only part of them. We think there are great
 possibilities in reproducing old pictures. We wish we had a Pickard group. Some
 Pickard descendant may wish to make a collection.

-We are much impressed with the future possibilities of getting a complete geneal-
+We are much impressed with the future possibilities of getting a complete geneol-[sic]
 ogy of the Pickard family. Mr. Cochran has a fine chapter on the Pickards but
 to date we have not had the pleasure of finding all of the family dates. We had
 intended to give more family data in this book but it takes time to get the
@@ -6,9 +6,7 @@ JOHN E. COCHRAN
 FAMILY ASSOCIATION
 Family Only
 Home | Sibling's Stories | 1st Cousins | Ancestor's Stories | Reunion History | Next Reunion | JECMEF
-
 OMIE WRITES HOME
-
 Ed. Note: The following letter was written by Omie Cochran in Nome, Alaska and sent to her
 sister, Ethel Shinn, in Canfield, Idaho in 1923. It has been stored away these 63 years in the
 original envelope with its 2 cent stamp. The letter has a number of references to the Shinn
@@ -17,7 +15,9 @@ Miss Saville was the nurse at the Nome Hospital that was mentioned in the articl
 the family newsletter two years ago.

 Nome Alaska August 26, 1923
+
 My Dear Ethel et al.
+
 I don't know when I did write or when you did
 but I am going to write now however and never
 the less. But I wish I could talk (I can yet but I
@@ -26,7 +26,7 @@ and Polly sit up and listen and that little black
 rascal of yours would fairly sparkle with
 listening. Can't I see him listening now to all the
 yarns we told last summer?
-[photo of people on ice with ship in background]
+[photo of people on ice with kayak and dog sled]
 You see, we-Miss Saville and I, took a trip north
 on the Buford and it was very interesting. We
 went north thru the Bering Strait into the Arctic and as far as the Ice Pack. There the captain
@@ -37,7 +37,6 @@ along the icebergs and the walrus were 'heisted' on board by the use of the cran
 tons of freight and the beasts were so huge that they made the pulleys just creak. They were
 over 12 feet long, as big around as three cows, had no feet but toenails on their flippers or
 flappers, no head but their body just suddenly ended with a hole for a mouth and big bristles
-
 all around it similar to a currycomb in coarseness; no ears but huge tusks of ivory. They are
 the most repulsive looking animals imaginable and tho I have always read about them I never
 expect such disagreeable looking creatures. They had a rough brown hairy skin and some of
@@ -80,7 +79,6 @@ us talk. Miss Saville talked quite a bit. Any how if you folks don't like this I
 all I had to write about and I know Buster'ud listen anyway and I'd soak ole Peter's head if he
 didn't and Polly would in my lap and I don't know much about the youngest one of yours so
 likely he would be squawling. But we did surely enjoy our trip and were gone just long enuf.
-
 I expect there were 150 passengers on board and almost or more of the crew and helpers. We
 had a stateroom down next to the kitchen and 'twas pretty fierce for odor at times.

@@ -2,30 +2,28 @@ source: Rod Moser Letter - p1.jpg
 provider: openrouter
 model: google/gemini-2.5-flash
 ---
-JOHN ISBILL
-R. T. MOSER
-
+JOHN ISBILL                                                                       R. T. MOSER
 ISBILL & MOSER
 DEALERS IN
 GENERAL MERCHANDISE

-Vonore, Tenn. January 27 – 1913
-Dear Uncle Aunt Adeline
+Vonore, Tenn. January 27 - 1913
+Dear Uncle [sic] Aun[t Adeline?]
 Was at home a
 few nights ago & saw a
-letter from your folks, So
+letter from you folks, so
 I decided to write you
-a few lines myself ok
-I am contemplate a
-trip out west next summer
-& [inserted: I] would like of adders [sic] to go
-where I [inserted: am] them.
+a few lines myself &
+I am contemplating a
+trip out west next summ[er]
+& I want both of fillers [sic] to go
+when I am [to] them.
 Am getting
-up in years & unmarried
+up in years & unmarried,
 so you see the object of
-my trip is to get a wife
-& if there is any old maid
-or widows out there, I
-want you to kiss them
-at my [inserted: mind] for me at there [sic]
-as soon as I get there.
+my trip, is to get a wife
+& I hear is a lot old maids
+& widows out there. I
+want you to see them
+at my land my [sic] at there [sic]
+as soon as I get there
@@ -5,7 +5,9 @@ isolated, fast, and leave no artifacts on disk.
 """

 import pytest
-from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel import Session
+from sqlmodel import SQLModel
+from sqlmodel import create_engine
 from sqlmodel.pool import StaticPool


@@ -6,7 +6,9 @@ import pytest
 from sqlmodel import select

 from transcription.config import Settings
-from transcription.models import Job, JobStatus, Transcript
+from transcription.models import Job
+from transcription.models import JobStatus
+from transcription.models import Transcript
 from transcription.providers.base import TranscriptionResult
 from transcription.services.upload import create_upload_job
 from transcription.worker import process_next_queued_job
@@ -71,6 +73,6 @@ class TestPipelineFailureFlow:
        assert job.status == JobStatus.FAILED
        assert transcript is not None
        assert transcript.text is None
-        assert "pipeline provider failure" in transcript.error_detail
+        assert "pipeline provider failure" not in transcript.error_detail
        assert "[internal_unexpected_error]" in transcript.error_detail
        assert "error_id=" in transcript.error_detail
@@ -5,8 +5,10 @@ from types import SimpleNamespace
 import pytest

 from transcription.config import Settings
-from transcription.providers.base import ProviderError, ProviderResponseError
-from transcription.providers.openrouter import DEFAULT_OPENROUTER_MODEL, OpenRouterTranscriptionProvider
+from transcription.providers.base import ProviderError
+from transcription.providers.base import ProviderResponseError
+from transcription.providers.openrouter import DEFAULT_OPENROUTER_MODEL
+from transcription.providers.openrouter import OpenRouterTranscriptionProvider


 class _FakeChat:
@@ -1,10 +1,18 @@
 """Tests for Step 3 library services (revisions, search, export)."""

-from sqlmodel import select
 import pytest
+from sqlmodel import select

-from transcription.models import Document, Job, JobStatus, Transcript, TranscriptRevision
-from transcription.services.library import accept_revision, add_revision, export_transcripts, list_revisions, search_accepted_transcripts
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import JobStatus
+from transcription.models import Transcript
+from transcription.models import TranscriptRevision
+from transcription.services.library import accept_revision
+from transcription.services.library import add_revision
+from transcription.services.library import export_transcripts
+from transcription.services.library import list_revisions
+from transcription.services.library import search_accepted_transcripts


 def _create_job(session) -> Job:
@@ -5,14 +5,13 @@ 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,
-)
+from transcription.providers.base import ProviderError
+from transcription.providers.base import TranscriptionResult
+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


 class _FakeProvider:
@@ -7,7 +7,6 @@ import pytest

 from transcription.services.transcription import transcribe_document_image

-
 HAS_OPENROUTER_KEY = bool(os.getenv("OPENROUTER_API_KEY"))

 REAL_IMAGES_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "real"
@@ -67,4 +66,4 @@ class TestRealImageExternalTranscription:
            f"{result.text}\n"
        )
        artifact_path.write_text(artifact_text, encoding="utf-8")
-        assert artifact_path.exists()
+        assert artifact_path.exists()
@@ -5,8 +5,11 @@ 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
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import JobStatus
+from transcription.services.upload import UploadError
+from transcription.services.upload import create_upload_job


@pytest.mark.unit
@@ -41,6 +44,24 @@ class TestUploadValidation:
        assert exc_info.value.category.value == "user_input_error"
        assert "jpg" in exc_info.value.suggestion.lower()

+    def test_rejects_payload_exceeding_max_upload_bytes(self, session, tmp_path: Path):
+        """create_upload_job rejects payloads above configured size limit."""
+        settings = Settings(
+            openrouter_api_key="test-key",
+            upload_dir=tmp_path,
+            max_upload_bytes=3,
+        )
+        with pytest.raises(UploadError) as exc_info:
+            create_upload_job(
+                filename="scan.jpg",
+                file_bytes=b"1234",
+                session=session,
+                settings=settings,
+            )
+
+        assert exc_info.value.category.value == "user_input_error"
+        assert "smaller file" in exc_info.value.suggestion.lower()
+

@pytest.mark.integration
 class TestUploadPersistence:
@@ -7,10 +7,16 @@ 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, TranscriptRevision
+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 transcription.models import Transcript
+from transcription.models import TranscriptRevision
 from transcription.providers.base import TranscriptionResult
-from transcription.worker import process_next_queued_job, run_worker_loop
+from transcription.worker import process_next_queued_job
+from transcription.worker import run_worker_loop


 def _create_queued_job(session, *, filename: str = "doc.jpg", file_path: str = "uploads/doc.jpg") -> Job:
@@ -118,7 +124,7 @@ class TestWorkerFailurePath:
        assert job.status == JobStatus.FAILED
        assert transcript is not None
        assert transcript.text is None
-        assert "provider failure" in transcript.error_detail
+        assert "provider failure" not 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
@@ -148,7 +154,7 @@ class TestWorkerFailurePath:
        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 "provider failure" not in transcripts[0].error_detail
        assert "[internal_unexpected_error]" in transcripts[0].error_detail
        assert "error_id=" in transcripts[0].error_detail

@@ -26,6 +26,8 @@ class TestAppLifespan:
        calls = []

        monkeypatch.setattr("transcription.app.setup_logging", lambda: calls.append("logging"))
+        monkeypatch.setattr("transcription.app.get_settings", lambda: object())
+        monkeypatch.setattr("transcription.app.enforce_request_access", lambda **_kwargs: None)
        monkeypatch.setattr("transcription.app.create_all", lambda **_kwargs: calls.append("schema"))
        monkeypatch.setattr(
            "transcription.app.initialize_database_runtime",
@@ -35,6 +37,8 @@ class TestAppLifespan:
        monkeypatch.setattr("transcription.app.should_bootstrap_schema", lambda _settings: True)
        monkeypatch.setattr("transcription.app._start_worker", lambda _app: calls.append("start_worker"))
        monkeypatch.setattr("transcription.app._stop_worker", lambda _app: calls.append("stop_worker"))
+        monkeypatch.setattr("transcription.app.apply_pending_migrations", lambda **_kwargs: calls.append("migrate"))
+        monkeypatch.setattr("transcription.app.validate_schema_compatibility", lambda **_kwargs: [])

        class _Dir:
            def mkdir(self, parents: bool, exist_ok: bool):
@@ -43,6 +47,8 @@ class TestAppLifespan:
        class _Settings:
            upload_dir = _Dir()
            prompt_dir = _Dir()
+            migration_auto_apply_on_startup = False
+            validate_schema_on_startup = True

        monkeypatch.setattr("transcription.app.get_settings", lambda: _Settings())

@@ -61,6 +67,8 @@ class TestAppLifespan:
        calls = []

        monkeypatch.setattr("transcription.app.setup_logging", lambda: None)
+        monkeypatch.setattr("transcription.app.get_settings", lambda: object())
+        monkeypatch.setattr("transcription.app.enforce_request_access", lambda **_kwargs: None)
        monkeypatch.setattr("transcription.app.create_all", lambda **_kwargs: None)
        monkeypatch.setattr(
            "transcription.app.initialize_database_runtime",
@@ -70,6 +78,8 @@ class TestAppLifespan:
        monkeypatch.setattr("transcription.app.should_bootstrap_schema", lambda _settings: True)
        monkeypatch.setattr("transcription.app._start_worker", lambda _app: calls.append("start_worker"))
        monkeypatch.setattr("transcription.app._stop_worker", lambda _app: calls.append("stop_worker"))
+        monkeypatch.setattr("transcription.app.apply_pending_migrations", lambda **_kwargs: calls.append("migrate"))
+        monkeypatch.setattr("transcription.app.validate_schema_compatibility", lambda **_kwargs: [])

        class _Dir:
            def mkdir(self, parents: bool, exist_ok: bool):
@@ -78,6 +88,8 @@ class TestAppLifespan:
        class _Settings:
            upload_dir = _Dir()
            prompt_dir = _Dir()
+            migration_auto_apply_on_startup = False
+            validate_schema_on_startup = True

        monkeypatch.setattr("transcription.app.get_settings", lambda: _Settings())

@@ -5,7 +5,8 @@ from pathlib import Path
 import pytest
 from pydantic import ValidationError

-from transcription.config import Provider, Settings
+from transcription.config import Provider
+from transcription.config import Settings


 def _make_settings(**overrides) -> Settings:
@@ -63,6 +64,33 @@ class TestPathSettings:
        assert isinstance(settings.prompt_dir, Path)


+class TestMigrationSafetySettings:
+    """Verify migration safety settings defaults."""
+
+    def test_migration_safety_defaults(self):
+        """Migration auto-apply is off and startup schema validation is on by default."""
+        settings = _make_settings()
+        assert settings.migration_auto_apply_on_startup is False
+        assert settings.validate_schema_on_startup is True
+
+
+class TestSecuritySettings:
+    """Verify Step 5 security-related settings behavior."""
+
+    def test_security_defaults(self):
+        """Security controls default to disabled auth and bounded upload size."""
+        settings = _make_settings()
+        assert settings.max_upload_bytes == 15 * 1024 * 1024
+        assert settings.operator_access_enabled is False
+        assert settings.operator_username == "operator"
+        assert settings.operator_password is None
+
+    def test_operator_password_required_when_access_enabled(self):
+        """Enabling operator access requires OPERATOR_PASSWORD."""
+        with pytest.raises(ValidationError):
+            _make_settings(operator_access_enabled=True, operator_password=None)
+
+
 class TestWorkerReliabilitySettings:
    """Verify worker retry settings defaults."""

@@ -1,7 +1,10 @@
 """Tests for transcription.db — schema bootstrap and session factory."""

-from sqlalchemy import inspect, text
-from sqlmodel import Session, SQLModel, create_engine
+from sqlalchemy import inspect
+from sqlalchemy import text
+from sqlmodel import Session
+from sqlmodel import SQLModel
+from sqlmodel import create_engine
 from sqlmodel.pool import StaticPool


@@ -18,12 +21,14 @@ class TestSchemaBootstrap:
    """Verify create_all produces the expected table set."""

    def test_create_all_creates_expected_tables(self):
-        """After create_all(), document, job, and transcript tables exist."""
+        """After create_all(), core V1 tables exist."""
        engine = _in_memory_engine()
        # Ensure models are imported so metadata is populated
-        from transcription.models import Document, Job, Transcript  # noqa: F401
-
        import transcription.db as db_module
+        from transcription.models import Document  # noqa: F401
+        from transcription.models import Job  # noqa: F401
+        from transcription.models import Transcript  # noqa: F401
+        from transcription.models import TranscriptRevision  # noqa: F401

        db_module.create_all(engine=engine)

@@ -32,6 +37,18 @@ class TestSchemaBootstrap:
        assert "document" in table_names
        assert "job" in table_names
        assert "transcript" in table_names
+        assert "transcriptrevision" in table_names
+
+    def test_validate_schema_compatibility_returns_no_issues_for_fresh_schema(self):
+        """validate_schema_compatibility reports no issues on fresh schema."""
+        engine = _in_memory_engine()
+
+        import transcription.db as db_module
+
+        db_module.create_all(engine=engine)
+        issues = db_module.validate_schema_compatibility(engine=engine)
+
+        assert issues == []


 class TestSessionFactory:
@@ -2,7 +2,10 @@

 import pytest

-from transcription.errors import AppError, ErrorCategory, classify_unexpected_error, new_error_id
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+from transcription.errors import classify_unexpected_error
+from transcription.errors import new_error_id


@pytest.mark.unit
@@ -38,6 +41,6 @@ class TestAppErrorHelpers:
        assert isinstance(err, AppError)
        assert err.category == ErrorCategory.INTERNAL_UNEXPECTED
        assert "unit.test" in err.message
-        assert "boom" in err.message
+        assert "boom" not in err.message
        assert err.suggestion
        assert err.error_id
@@ -0,0 +1,108 @@
+"""Tests for transcription.migrations — explicit Step 4 migration safety behavior."""
+
+from sqlalchemy import inspect
+from sqlalchemy import text
+from sqlmodel import create_engine
+from sqlmodel.pool import StaticPool
+
+from transcription.migrations import apply_pending_migrations
+from transcription.migrations import list_pending_migrations
+
+
+def _in_memory_engine():
+    """Create isolated in-memory SQLite engine."""
+    return create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+
+
+class TestMigrations:
+    """Verify migration listing and application behavior."""
+
+    def test_list_pending_returns_all_before_apply(self):
+        """All known migrations are pending on a fresh legacy-shaped database."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        pending = list_pending_migrations(engine=engine)
+        assert [migration.revision_id for migration in pending] == [
+            "0001_add_retry_count_to_job",
+            "0002_create_transcriptrevision_table",
+        ]
+
+    def test_apply_pending_migrations_records_history_and_schema(self):
+        """Applying pending migrations mutates schema and records revision history."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        applied = apply_pending_migrations(engine=engine)
+        assert applied == [
+            "0001_add_retry_count_to_job",
+            "0002_create_transcriptrevision_table",
+        ]
+
+        inspector = inspect(engine)
+        job_columns = {column["name"] for column in inspector.get_columns("job")}
+        assert "retry_count" in job_columns
+        assert "transcriptrevision" in set(inspector.get_table_names())
+
+        with engine.begin() as connection:
+            rows = connection.execute(
+                text("SELECT revision_id FROM schema_migration_history ORDER BY revision_id")
+            ).fetchall()
+        assert [row[0] for row in rows] == applied
+
+    def test_apply_pending_migrations_is_idempotent(self):
+        """Re-running apply_pending_migrations with no pending revisions is a no-op."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        first_apply = apply_pending_migrations(engine=engine)
+        second_apply = apply_pending_migrations(engine=engine)
+
+        assert len(first_apply) == 2
+        assert second_apply == []
@@ -5,7 +5,11 @@ from uuid import UUID
 import pytest
 from sqlalchemy.exc import IntegrityError

-from transcription.models import Document, Job, JobStatus, Transcript, TranscriptRevision
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import JobStatus
+from transcription.models import Transcript
+from transcription.models import TranscriptRevision


 def _make_document(**overrides) -> Document:
@@ -2,7 +2,6 @@

 from pathlib import Path

-
 PROMPT_PATH = Path("prompts/transcribe_document.md")


@@ -4,8 +4,11 @@ from uuid import uuid4

 import pytest

-from transcription.models import Document, Job, Transcript
-from transcription.ui.jobs_page import fetch_job_detail, fetch_jobs
+from transcription.models import Document
+from transcription.models import Job
+from transcription.models import Transcript
+from transcription.ui.jobs_page import fetch_job_detail
+from transcription.ui.jobs_page import fetch_jobs


@pytest.mark.integration
@@ -1,8 +1,8 @@
 """Tests for UI page registration wiring."""

+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
-import pytest

 from transcription.ui import register_pages

@@ -5,7 +5,8 @@ from uuid import uuid4

 import pytest

-from transcription.services.upload import UploadError, UploadJobResult
+from transcription.services.upload import UploadError
+from transcription.services.upload import UploadJobResult
 from transcription.ui import upload_page
Author	SHA1	Message	Date
John Lancaster	cf5d7d8c7b	Merge branch 'main' into diverging	2026-06-26 18:35:27 -05:00
Jim Lancaster	939b0e46e9	Merge branch 'main' of https://gitea.john-stream.com/bbchops/transcription	2026-06-26 18:19:14 -05:00
Jim Lancaster	15af11ecb5	ver1 - Step6 Implementation Plan	2026-06-26 18:19:12 -05:00
John Lancaster	6dc58a8d50	ruff linting	2026-06-26 18:19:09 -05:00
Jim Lancaster	e90dbe4958	ver1 - Step 5 implementation complete.	2026-06-26 17:51:02 -05:00
Jim Lancaster	c9682b0399	ver1 - Step 5 implementation plan	2026-06-26 15:25:57 -05:00
Jim Lancaster	dd80cd60cf	ver1 - Step 4 implemented - Revision functionality added	2026-06-26 14:59:16 -05:00
Jim Lancaster	754a273ead	Ver1 Step4 implementation plan	2026-06-26 14:37:30 -05:00
John Lancaster	5ef74ef33a	breaking up pages	2026-06-26 01:00:38 -05:00
John Lancaster	e4889ba584	ruff and settings	2026-06-26 00:26:35 -05:00