Phase 6: enforce the quality gate in CI and export the V4.7 review log

Adds .github/workflows/quality-gate.yml, running the gate on push and pull request. CI invokes `pre-commit run --all-files` rather than restating the `ruff check` and `ty check` commands, so the checks keep a single definition in .pre-commit-config.yaml and local and CI cannot drift (plan task 2). The workflow writes a .env file rather than exporting an environment variable. The two are not equivalent here: Settings reads the .env file, while the external-test skip guard reads os.getenv, so an exported variable un-skips the external tests and sends them to the network. Measured in CI: no .env gave 115 failures and 18 errors, an exported dummy key gave 3 failures, and a written .env file reproduced the local baseline exactly. Negative-tested on a scratch branch: a deliberate lint error failed the run at `ruff check` with exactly the planted errors, confirming the gate blocks rather than merely reporting (plan task 4). The subsequent clean run passed ruff and ty and reported 295 passed, 4 skipped, matching local and confirming the four credential-gated tests skip cleanly (plan task 3). That first green run caught a real platform-dependent defect. PromptStore rejected non-direct-child names via `Path(name).name != name`, which is platform-dependent: on POSIX a backslash is an ordinary filename character, so "nested\prompt.md" passed the guard and failed later as NOT_FOUND rather than VALIDATION. Windows cannot reproduce it. No traversal was possible, since the path.parent != root check still held, so the impact was a wrong error category and a red gate. Both separators are now rejected explicitly, matching the ^[^/\\]+$ pattern config.PromptFilename already used. Also exports docs/ver4.7/review_log_v4_7.md, the working record kept across all six phases: 50 entries, 1 still open. The open entry is a pre-existing /ui redirect defect found during the Phase 3 UI walk and deliberately left unfixed as outside the V4.7 scope boundary. Co-authored-by: Copilot App <[email protected]>
Phase 5: contain worker faults instead of discarding the classification
2026-08-18 23:05:52 -05:00 · 2026-08-18 16:12:52 -05:00 · 2026-08-18 16:02:58 -05:00 · 2026-08-18 15:54:27 -05:00
21 changed files with 1103 additions and 251 deletions
@@ -7,29 +7,89 @@ applyTo: 'src/transcription/services/*.py'
 ## Structure
- Project core data models defined in [models](../../src/transcription/db/models.py)
+- Project core data models are defined in [models](../../src/transcription/db/models.py)
- 1 service class per data model
+- One service class per **aggregate**, not per table. An aggregate is a root model plus
- Only services directly interact with the database, and only through async methods
+  the models that have no independent lifecycle of their own. `DocumentType` has no
- Services are completely independent of one another. Any operation that needs to use more than a single service, which is most of them, needs to have a separate orchestration function.
+  meaning without `Document`, so it belongs to `DocumentService`; it does not get its
  own service. Splitting per table produces services that must reach across each other
  for every real operation, which is what line 13 forbids.
 - Only services interact with the database, and only through async methods.
 - **A service module must not import another service module.** This is enforced by
  [test_service_boundaries](../../tests/test_service_boundaries.py). Shared types go in a
  neutral module that defines no service class (see [errors](../../src/transcription/services/errors.py)).
 - Not every module in this package is a service. Helper modules that define no `*Service`
  class (`base`, `errors`, `normalization`, `prompts`, `quality`, `media_storage`,
  `source_media`) are free-function modules and are exempt from the service rules below.
 ## Model Ownership
 Every model has exactly one owning service. The owner defines that model's invariants and
 is the only service that may **create or delete** its rows.
 | Model | Owner |
 | --- | --- |
 | `Document`, `DocumentType` | `DocumentService` |
 | `Source`, `JobSource` | `SourceService` |
 | `Job` | `JobService` |
 | `Person`, `PersonRole`, `DocumentPerson` | `PeopleService` |
 | `ExecutionAttempt` | `EvidenceService` |
 ### Junction tables
 A junction table is owned by the service that **creates and deletes its rows** — its
 lifecycle owner. The service on the other side may read through the junction (via
 `selectinload`) but must not create rows in it.
 - `document_person` -> `PeopleService`. Every write is there; `DocumentService` only
  eager-loads through it.
 - `job_source` -> `SourceService`, which creates the row, records each page's outcome,
  and deletes it.
 Two consequences follow, and both are deliberate:
 - **Cascade deletion is not a violation.** A service deleting the aggregate root it owns
  may delete junction rows referencing that root, because they cannot outlive it
  (`JobService.delete_job_with_guardrails`).
 - **Ownership governs creation and deletion, not every state transition.** `job_source` is
  both a link and the transcription work queue. `JobService.cancel_job` and
  `resubmit_failed_sources` transition `job_source.status` across a whole job, because that
  transition is a Job lifecycle event, not a per-page outcome. They create and delete
  nothing.
 `EvidenceService.promote_machine_attempt` writes two fields on `Source`
 (`preferred_execution_attempt_id`, `raw_transcription`). This is allowed on the same
 principle: selecting which attempt a Source presents is an evidence decision that happens
 to land on `Source`. It is scoped to those two projection fields.
 If a new operation cannot be expressed within one owner, it belongs in an orchestration
 module, not in a cross-service import.
 ## Error Handling
- Service-specific errors defined at the top of the respective module and inherit from `AppError`
+- Errors used by a single service are defined at the top of that module and inherit from `AppError`.
- Use a context manager for large `try/except` blocks like `handle_transcription_errors` in [sources](../../src/transcription/services/sources.py)
+- Errors shared by more than one service go in [errors](../../src/transcription/services/errors.py),
  which defines no service class and is therefore importable by any of them.
 - Use a context manager for large `try/except` blocks, like `handle_transcription_errors` in
  [sources](../../src/transcription/services/sources.py).
 ## Checklist
 - [ ] Uses `ServiceBase` for common logic
- [ ] CRUD methods created at the top
+- [ ] Session kwarg for `AsyncSession` to pass a session object into each method
- [ ] Session kwarg for `AsyncSession` to pass in a session object to each method
+- [ ] Services use `self._session_scope` in their methods to pass the session through
- [ ] Services use `self._session_scope` in their methods to pass the session thru.
+    - Multiple operations on the same object(s) require sharing a session between all the methods used
-    - Multiple operations on the same object(s) require sharing a session between all the methods used.
+- [ ] Every model the module touches is either owned by it or reached read-only
 ## CRUD Methods
- Create, read, update, and delete, created in that order
+- Name format `<operation>_<model>`, for example `create_document` or `update_job`.
- Name format `<operation>_<model >`, for example `create_document` or `update_job`
+- Where a service exposes create/read/update/delete for its root model, define them at the
- All services must define these 4 methods first, and in that order
+  top of the class in that order, before derived reads and workflow helpers.
 - Not every aggregate needs all four. `ExecutionAttempt` is append-only evidence written by
  `workflows.py`, so `EvidenceService` deliberately exposes reads and no create or delete.
  Do not add unused CRUD methods to satisfy symmetry.
 - `RegistryService` is generic across small lookup models and uses `<operation>_entry`
  naming instead.
 ## Transaction Finalization
@@ -74,4 +134,11 @@ Separation of concerns:
 # Service Composition
-Some operations, like uploading a picutre, require modifications to multiple tables, which can be done by composing methods from the service object into a separate function.
+A service method may read across models it does not own, using eager loads from its own
 aggregate root. What it may not do is import another service.
 Operations that must **write** models owned by more than one service — uploading a picture,
 for example — are composed in an orchestration module
 ([store](../../src/transcription/services/store.py),
 [workflows](../../src/transcription/services/workflows.py)). Orchestration modules define no
 service class, may import any service, and own the commit boundary.
@@ -0,0 +1,45 @@
 name: Quality Gate
 # V4.7 Phase 6 / review log [40]. Before this, ruff, ty and pytest were enforced
 # only by .pre-commit-config.yaml, and only for developers who had actually run
 # `pre-commit install`.
 on:
  push:
  pull_request:
 jobs:
  gate:
    runs-on: ubuntu-latest
    steps:
      - name: Check out the commit under test
        uses: actions/checkout@v4
      - name: Install uv
        run: |
          curl -LsSf https://astral.sh/uv/install.sh | sh
          echo "$HOME/.local/bin" >> "$GITHUB_PATH"
      - name: Install dependencies from the lockfile
        # --locked fails if uv.lock has drifted from pyproject.toml, so a stale
        # lockfile is caught here rather than producing an untested dependency set.
        run: uv sync --locked
      - name: Write placeholder configuration
        # Settings requires openrouter_api_key and 115 tests cannot construct
        # Settings without it. This is written to a .env file rather than exported
        # as an environment variable on purpose: the external tests guard on
        # os.getenv("OPENROUTER_API_KEY"), which reads the process environment and
        # not the file, so writing the file reproduces the local result exactly -
        # the 4 external tests skip instead of running against a fake key and
        # failing. Exporting it instead produces 3 failures.
        run: echo "OPENROUTER_API_KEY=ci-placeholder-not-a-real-key" > .env
      - name: Lint and type check
        # Runs the hooks defined in .pre-commit-config.yaml instead of repeating
        # "ruff check" and "ty check" here. The commands then have one definition,
        # so the local and CI gates cannot drift apart.
        run: uv run pre-commit run --all-files --show-diff-on-failure
      - name: Tests
        run: uv run pytest
@@ -0,0 +1,415 @@
 # V4.7 Implementation Review Log
 Working record kept during the V4.7 architectural cleanup release.
 This file is the canonical reference for citations of the form **`review log [N]`** in V4.7 and later planning documents. The numbers below are those `N` values. They are independent of the [V4.6 log](../ver4.6/review_log_v4_6.md), which has its own numbering.
 The log was maintained live in a session-scoped database and exported here so the citations remain resolvable in later sessions. It is a historical record: entries are not rewritten after the fact, except where a later phase resolved an entry that was open at the time, in which case the resolution is appended to the body and marked `RESOLVED:`. Where an entry conflicts with a committed planning document, **the planning document wins**.
 ## Legend
 | Field | Meaning |
 | :--- | :--- |
 | `kind` | `question` - needed a decision; `comment` - observation; `deviation` - departure from plan; `risk` - identified hazard |
 | `status` | `open` - unresolved; `answered` - resolved by a decision; `noted` - recorded, no action required |
 | `finding` | Finding ID in [architecture_code_review_2026-08-17.md](../architecture_code_review_2026-08-17.md), where one applies. Most V4.7 entries have none, because V4.7 works from the [implementation plan](implementation_plan_v4_7.md) rather than from that pre-V4.6 snapshot. Where an entry carries a one-line summary instead, it appears as a bold lead-in to the body. |
 **50 entries** - 1 open, 18 answered, 31 noted.
 ## Still Open
 These carry forward past V4.7.
 | ID | Finding | Summary | Disposition |
 | :--- | :--- | :--- | :--- |
 | [32] | - | /ui/documents/{id}/sources redirects to /sources, dropping the /ui prefix | Pre-existing and outside the V4.7 scope boundary - deliberately left unfixed |
 ## Full Log
 ### Phase 0 - baseline and backups
 #### [1] Backups taken and verified
 *comment* - **noted**
 data/transcription.db and data/documents/ copied to C:\GitHub\_backups\transcription_v47_20260818-092616. All 76 document files SHA256-identical to source. A consistent SQLite snapshot (transcription.consistent.db) was also produced via the sqlite3 backup API because the live DB file is locked by a running app process, making a plain file copy potentially torn.
 #### [2] The application appears to be running and holds data/transcription.db
 *risk* - **answered**
 Two python processes started 2026-08-18 04:58 (PID 14340 is .venv python). Get-FileHash on data/transcription.db failed with a sharing violation. The Phase 1 migration rewrites data/documents/ JPEGs in place and later phases ALTER the live schema; both are unsafe while the app is running. Need the user to stop the app before any migration run. RESOLVED: the user stopped the app before the migration ran; the DB accepted an immediate write lock beforehand.
 #### [3] Working tree is clean; the V4.7 doc edits are already committed
 *deviation* - **noted**
 The brief said both V4.7 docs have uncommitted edits on disk. git status --short is empty and HEAD is 246d7f9 "V4.7 final scope changes", which contains them. Nothing was reverted or stashed; the working tree content matches what the brief described.
 #### [4] Baseline is not fully green: 1 pre-existing test failure caused by .env leakage
 *risk* - **answered**
 tests/test_config.py::TestWorkerReliabilitySettings::test_worker_retry_defaults asserts worker_max_retries == 0 but the local .env sets WORKER_MAX_RETRIES=1, and _make_settings() reads .env. ruff and ty are both clean. This is a test-isolation defect predating V4.7 (introduced when the 2026-08-18 calibration wrote WORKER_MAX_RETRIES into .env), not a code defect. It would pass in CI, where no .env exists, which makes it a latent local/CI divergence relevant to Phase 6. Awaiting a decision on whether to fix it. RESOLVED: tests/test_config.py::_make_settings now builds Settings with _env_file=None, isolating it from the local .env. The local/CI divergence this entry flagged is gone, and CI run 182 confirms the same 295 passed / 4 skipped result on Linux with no .env of the developer's.
 ### Phase 1 - artifact removal and ingest rotation
 #### [5] Where does ingest-time rotation actually belong: media_storage or store.py?
 *question* - **answered**
 Plan Phase 1 task 1 says rotate inside media_storage ahead of write_bytes. But store.py computes file_hash and file_size_bytes from the ORIGINAL file_bytes (store.py:93, 168) independently of write_media_bytes, which returns only a Path. Rotating inside write_media_bytes would leave Source.file_hash and file_size_bytes describing bytes that were never stored. write_media_bytes is also shared with person portraits and homepage images. Recommend rotating the bytes once at the Source-ingest boundary (store_source_file or its two callers in store.py) so hash, size and stored file all describe the same upright bytes, and leaving media_storage a generic byte writer. [ANSWERED 2026-08-18 by user] Rotate at the Source-ingest boundary, before file_hash/file_size_bytes are computed, so the hash and size describe the stored upright bytes. media_storage stays a generic byte writer. This is a deliberate deviation from Phase 1 task 1 as written.
 #### [6] Fate of artifact_dir / artifact_inline_threshold_bytes settings and data/artifacts/
 *question* - **answered**
 Removing ProcessingArtifact orphans Settings.artifact_dir and Settings.artifact_inline_threshold_bytes (config.py:104,106), the JobService artifact deletion path (jobs.py:307-350), and the on-disk data/artifacts/ tree. config.py is outside the services instruction file. Proposal: delete both settings and the jobs.py deletion path as part of the same removal, and have the migration delete the external artifact files (migration step 2 already says so). Confirm. [ANSWERED 2026-08-18 by user] Delete both Settings.artifact_dir and Settings.artifact_inline_threshold_bytes.
 #### [7] Phase 1 blast radius is wider than the plan task list
 *comment* - **noted**
 Beyond the listed sites, ProcessingArtifact is also referenced by: sources.py delete guards (lines 296-300 and 450-462, which block Source deletion when artifacts exist), jobs.py job-deletion artifact cleanup (307-350), Source.processing_artifacts and ExecutionAttempt.artifacts relationships (models.py:322,431), the selectinload(ExecutionAttempt.artifacts) in list_execution_attempts (sources.py:719), and the ProcessingArtifact lookup that validates model_input_artifact_id inside update_job_source_transcription (sources.py:648-664). All must go with the model.
 #### [10] Quality-warnings render: delete (per plan) or re-point at normalized_metadata?
 *question* - **answered**
 Plan Phase 1 task 7 says delete the render at sources_page.py:661, but task 4 folds the same payload into execution_attempt.normalized_metadata and decision A forbids user-facing change. The render has never fired in practice because it reads attempt.artifacts and only 2 artifact rows exist. [ANSWERED 2026-08-18 by user] Keep the display and re-point it at normalized_metadata. Deviation from Phase 1 task 7 as written; task 4 now has a consumer.
 #### [11] SourceEvidenceReference.derivative_id / transformation kept but no longer populated
 *deviation* - **noted**
 With ProcessingArtifact gone there is no derivative to reference, so both fields are always None. They were left in place rather than removed: RequestManifest is a frozen, versioned evidence contract whose canonical bytes feed request_manifest_sha256, so removing fields would change the digest of every future manifest and arguably require a schema_version bump - cost out of proportion to deleting two optional fields. Raise if you would rather see the contract cleaned up.
 #### [12] The session.commit() at workflows.py:228 is removed with resolve_provider_input
 *deviation* - **noted**
 That commit existed to make the artifact row written during provider-input resolution durable before the provider call. With no artifact write there is nothing pending to commit - the PROCESSING claim was already committed at line 198 / 426 - so the call is removed rather than left as a no-op. This also removes one of the two things Phase 4 has to get out of the duration measurement window.
 #### [13] The image backfill must also update source.file_hash and file_size_bytes
 *risk* - **answered**
 Migration step 1 as written only rotates the stored JPEGs and strips the EXIF tag. But source.file_hash and source.file_size_bytes were computed from the pre-rotation bytes, and after Phase 1 the transcription path derives the evidence digest (SourceEvidenceReference.digest_sha256) straight from source.file_hash. Rotating the file without updating the row would make every backfilled Source advertise a digest that does not match the bytes actually sent to the provider - the exact class of defect the evidence model exists to prevent. The migration therefore rewrites both columns for each rotated image in the same transaction. Not a change of intent, an omission in the step description.
 #### [14] Orientation normalization must not change what ingest accepts
 *deviation* - **noted**
 **Undecodable upload bytes are a normalization no-op, not a rejection**
 validate_source_content only checks emptiness and filename; it never decoded the image, so bytes that Pillow cannot open (e.g. the b"image-bytes" fixture in tests/services/test_store.py) were accepted and stored. Moving rotation into store_source_file initially turned that into an OrientationNormalizationError, i.e. a user-facing rejection of previously accepted uploads. Decision A forbids user-facing change, so Image.open failure now logs and returns None; the error is retained only for a decode that succeeded and a rewrite that then failed.
 #### [15] Artifact-only tests removed with the subsystem
 *deviation* - **noted**
 **Two tests deleted rather than rewritten**
 tests/test_v42_evidence.py::test_large_json_artifact_uses_constrained_atomic_storage and ::test_rejects_inline_artifact_with_incorrect_integrity exercised only external artifact storage and inline artifact integrity. Both behaviours are deleted by Phase 1, so the tests have no surviving subject. tests/ui/test_sources_page.py lost one assertion ("Derived Artifacts"), and tests/test_db.py lost the processing_artifact table assertion.
 #### [16] Fate of the superseded v4.5 to v4.6 migration tool
 *question* - **answered**
 **tools/migrate_v45_to_v46.py no longer type-checks**
 The old migration references ProcessingArtifact (line 157), SourceService._verify_artifacts_integrity (line 169), and carries processing_artifact: 2 in EXPECTED_SOURCE_COUNTS (line 83). All three are gone. It is currently the only remaining ty failure. The plan says to keep its enum-spelling tolerance but does not address this. Options: delete the completed one-time tool; or strip the artifact code path from it. RESOLVED: the completed one-time tool was deleted (Phase 1). tools/ now contains only migrate_v46_to_v47.py, and ty is clean.
 #### [17] tools/migrate_v45_to_v46.py removed rather than repaired
 *question* - **answered**
 **Superseded v4.5 to v4.6 migration deleted**
 User decision. The migration is complete, the live database is already V4.6, and after V4.7 it would restore a V4.5 backup into a schema that no longer matches (job_source is stripped in Phase 2). Two doc references remain, both citing it only as a conventions template: implementation_plan_v4_7.md lines 39 and 50, scope_boundary_v4_7.md line 160. Line 50 (enum-spelling tolerance) is now moot. Recoverable from git history if ever needed.
 #### [18] DEFAULT_ARTIFACT_DIR constant in migrate_v46_to_v47.py
 *deviation* - **noted**
 **Migration records the deleted artifact_dir default itself**
 Step 2 must delete external artifact files, but Settings.artifact_dir was deleted in the same phase. The migration therefore carries the historical V4.6 default (data/artifacts) as its own constant with a --artifact-dir override, rather than depending on a setting that no longer exists. One external file was present and removed; the directory is now empty.
 #### [19] Phase 1 migration outcome
 *risk* - **answered**
 **Migration executed and verified against the live corpus**
 Ran after the user stopped the app and after a fresh pre-migration backup to C:\GitHub\_backups\transcription_v47_premigration_20260818-101232. Result: 58 images rotated, 18 already upright, 0 missing; processing_artifact dropped (2 rows) and its 1 external file removed. Verification: no stored image reports orientation 3/6/8; source.file_hash and file_size_bytes match every file on disk (0 mismatches over 76); 58 of 76 files differ from the backup; PSNR against the un-rotated backup is 50.3 / 51.1 / 56.1 dB (min/median/max) across the 57 JPEGs, allowing for the -6 percent size reduction; a re-run reports rotated=0 and table already absent, confirming idempotency. Visual spot-check of 1547e555 confirmed the page was genuinely stored upside down and is now upright.
 ### Phase 2 - evidence model simplification
 #### [8] Should CANCELLED pages be re-attempted when a job is re-run?
 *question* - **answered**
 Plan Phase 2 task 7. _resolve_job_sources (workflows.py:432-442) selects work by status != TRANSCRIBED, so once CANCELLED exists as a distinct status a re-run would silently pick cancelled pages back up. Options: (a) exclude CANCELLED from work selection, so cancelling is sticky and a page must be explicitly re-queued; (b) include it, so re-running a job means "do everything not yet transcribed"; (c) clear CANCELLED back to PENDING in the existing retry path (jobs.py:411-424) and exclude it from work selection, which makes re-attempt an explicit user action through the retry button. Decision required before Phase 2 task 1. RESOLVED: re-attempt them. Resubmit accepts FAILED and CANCELLED. Rationale: today cancel writes FAILED, so resubmit already resets cancelled pages to PENDING; introducing a distinct CANCELLED status without widening the resubmit filter would silently make cancelled work unrecoverable, a user-facing regression that decision A forbids. The decision is encoded in the resubmit candidate filter, which is the real decision point - _resolve_job_sources only ever sees these rows after resubmit has already set PENDING. UI copy on both the cancel and resubmit pages is updated to match.
 #### [20] Plan task 4 targets a dead module
 *question* - **answered**
 **ui/components/transcript.py deleted instead of redirected**
 Phase 2 task 4 directs transcript.py:103-119 to sort by ExecutionAttempt.finished_at instead of job_source.executed_at. Investigation showed the module is entirely unreferenced: no import of transcription.ui.components.transcript exists in src, tests, or docs, and both public functions (render_original_transcription_card, render_revision_row) have zero callers. Rewriting it would mean maintaining unreachable code against the new evidence model. User decision: delete the module. Recoverable from git history.
 #### [21] Defect [45] fixed by declaring one enum spelling
 *deviation* - **noted**
 **execution_attempt.status gains values_callable**
 ExecutionAttempt.status was a bare JobSourceStatus annotation, so SQLAlchemy persisted enum names (TRANSCRIBED) while job_source.status persisted values (transcribed) via values_callable. That is why the two columns matched on 0 of 79 rows. execution_attempt.status now declares the identical SAEnum with values_callable and native_enum=False. Existing rows carry the old spelling and are rewritten by migration step 3.
 #### [22] Dead property made more expensive by the evidence move
 *question* - **answered**
 **Job.error_detail deleted rather than re-derived**
 Plan task 4 lists models.py:266-278 (Job.error_detail) for redirection. A full-repo search found zero readers: JobTableRow has no such field and the job detail page never calls it. Re-deriving it from ExecutionAttempt would require a two-level eager load (job_sources -> execution_attempts) on every Job, across a lazy=raise then lazy=noload chain, where a missing load returns an empty list and the property would silently answer None instead of raising. No information is lost: error_detail survives on ExecutionAttempt and is reachable via list_execution_attempts and read_latest_execution_attempt. A future job-level failure view should query attempts directly anyway, since first-error-across-pages is the wrong shape for a partial-success job. User decision: delete.
 #### [23] Replacement ordering key after executed_at is dropped
 *question* - **answered**
 **Source.latest_job_source orders by Job.date_created**
 JobSource retains only id, job_id, source_id and status, so max(job_sources, key=executed_at) needs a key from a neighbour. Job.date_created is chosen over the latest ExecutionAttempt.finished_at: it is always present (a PENDING page has no attempt at all), it is already eager-loaded by read_source_detail, and since (job_id, source_id) is unique per source the ordering is exactly most recent job. The two differ only when a job created earlier finishes later, which the single-worker queue does not produce. User decision.
 #### [24] The "Cancelled by user" string has no home after job_source is stripped
 *deviation* - **noted**
 **Cancel no longer records a reason string**
 cancel_job previously wrote error_detail="Cancelled by user" onto job_source. That column is gone, and cancel deliberately makes no provider call so it writes no ExecutionAttempt. The reason is now carried by JobSourceStatus.CANCELLED itself, which is strictly more precise than a free-text string. UI copy on the cancel page was updated to say "cancelled" and to state that cancelled sources can be resubmitted.
 #### [25] jobs_page "Failed Sources" became "Resubmittable Sources"
 *deviation* - **noted**
 **Resubmit UI counter renamed**
 The resubmit candidate filter now accepts FAILED and CANCELLED per the user decision in entry 8, so the page counter had to count both. Renamed the metadata row and the blocked-error message accordingly.
 #### [26] sources_page no longer renders ai_metadata/raw_api_response when no attempt exists
 *comment* - **noted**
 **Legacy job_source evidence fallback deleted from the detail page**
 The "no ExecutionAttempt" branch of _render_provider_evidence used to fall back to the job_source JSON columns for historical rows. Those columns are gone, so the branch now renders only the empty state. Verified against the evidence baseline: all 77 successful transcriptions have a matching execution_attempt row, so no live row loses its evidence display.
 #### [27] latest_error_detail reads through job_sources -> execution_attempts
 *risk* - **noted**
 **Model properties now require a two-level eager load**
 Source.latest_error_detail feeds a visible "Error Detail" column on the sources table. Because JobSource.execution_attempts is lazy="noload" it returns empty rather than raising when not loaded, so a caller that forgets the chained selectinload gets a silent blank instead of an error. list_sources_detail and the model-property test were both updated to chain selectinload(...).selectinload(orm_attribute(...)). Any new caller must do the same.
 #### [28] job_source.status and execution_attempt.status now agree on every row
 *comment* - **noted**
 **Defect [45] verified fixed against the live database**
 Before: 0/79 rows matched, because execution_attempt persisted enum names and job_source persisted values. After migration step 3: 79/80 join rows agree. The single disagreement is job_source 09cd5f77 which has two attempts - attempt 1 failed, attempt 2 transcribed - so the queue row correctly reflects the final outcome while the history preserves the failure. Comparing job_source against its LATEST attempt gives 79/79.
 #### [29] list_sources_detail resolves latest_status and latest_error_detail for all 76 rows
 *comment* - **noted**
 **Two-level eager load verified against live data, not just tests**
 Ran SourceService.list_sources_detail against the migrated production database: 76 sources, 75 transcribed / 1 failed, and the one failed row still exposes latest_error_detail - now read from execution_attempt rather than the dropped job_source column. This closes the silent-blank risk recorded in entry 27 for the shipped call path.
 ### Phase 3 - evidence service extraction and the ownership rule
 #### [9] Junction ownership: which service owns job_source and document_person?
 *question* - `MED-14` - **answered**
 services.instructions.md names four core components (Document, Source, Job, Person) and is silent on the two junctions, which is exactly where two owners intersect. Candidate tie-break rules: (a) the junction belongs to the service that creates its rows; (b) it belongs to the aggregate whose lifecycle it shares (job_source dies with the Job, document_person dies with the Document); (c) it belongs to the side that reads it most. These do not agree for job_source: it is created by store.py orchestration, its lifecycle is the Job, and it is read predominantly through Source pages, which is how it ended up in sources.py. Decision required at Phase 3 task 7. RESOLVED: measurement showed document_person has a single writer (people.py, every create/delete/sync) and needs no tie-break; documents.py only eager-loads through it. job_source is genuinely contested between sources.py (row existence + per-page outcome) and jobs.py (job-lifecycle status transitions). User selected the LIFECYCLE rule: the service that creates and deletes rows owns the junction, so job_source -> SourceService. Two scoped carve-outs written into the rule: (1) cascade deletion of junction rows when a service deletes its own aggregate root (JobService.delete_job_with_guardrails); (2) status transitions that create and delete nothing (cancel_job, resubmit_failed_sources), because those are Job lifecycle events. No code was moved.
 #### [30] Where should the shared transcription error hierarchy live?
 *question* - **answered**
 **Extraction immediately violated the existing no-sibling-import rule**
 tests/test_service_boundaries.py enforces services.instructions.md:13 - a service module must not import a sibling. evidence.py needed TranscriptionNotFoundError, which sources.py also raises, so the extraction failed the rule on the first run. Measured ownership: CandidatePromotionError is now raised only in evidence.py; PromptLoadError and SourceDeleteBlockedError only in sources.py; TranscriptionNotFoundError in both; TranscriptionError is the shared base, caught by store.py. User chose to move the whole five-class hierarchy to a neutral services/errors.py: one obvious home, one import path, and the exception a caller catches no longer changes when an operation moves between services.
 #### [31] Two test bundles broke on adding a fifth service, not on the refactor itself
 *comment* - **noted**
 **ServiceBundle default factories silently bind to the real database**
 test_v45_candidates and test_workflows_reliability constructed ServiceBundle(...) field by field. Adding the evidence field meant it fell back to field(default_factory=EvidenceService), which resolves the process-global session factory rather than the test one - so the tests silently queried the wrong database instead of failing loudly. Both were changed to ServiceBundle.from_session_factory(...), which is immune to future additions. This is the same global-singleton hazard recorded in the 2026-08-17 review at line 272.
 #### [32] /ui/documents/{id}/sources redirects to /sources, dropping the /ui prefix
 *risk* - **open**
 **Pre-existing broken redirect found during the UI walk**
 The Phase 3 exit criterion requires walking every /ui/* page. 24 of 25 routes return 200. documents_page.py returns RedirectResponse(url=f"/sources?document_id=...") without the /ui mount prefix, so following the 307 lands on a 404. Confirmed pre-existing: documents_page.py has no uncommitted diff and was last touched in 6a3ee26, well before V4.7. Out of the V4.7 scope boundary, so NOT fixed - raised for the user to decide.
 #### [33] Instruction-file defects corrected
 *comment* - **noted**
 **services.instructions.md rewritten after the decomposition, per the mandated order**
 All five defects from plan Phase 3 task 7 fixed. (a) Line 11 "1 service class per data model" replaced with one service class per AGGREGATE, with DocumentType-under-DocumentService as the worked example; this is the measured cause of sources.py reaching 1,389 lines. (b) Added a Model Ownership section with a table covering every model plus an explicit junction-table rule, which the file previously had no home for. (c) The mandatory-CRUD rule (old lines 30-32) was already false: prompts.py, quality.py, normalization.py, media_storage.py and source_media.py define no service class at all, EvidenceService deliberately exposes no create/delete because ExecutionAttempt is append-only, and RegistryService uses generic <op>_entry naming. Softened to intent plus an explicit "do not add unused CRUD to satisfy symmetry". (d) Old line 13 (services fully independent) read as contradicting old lines 75-77 (compose across tables); reworded to separate READING across models via eager loads from the owning root, which is allowed, from IMPORTING another service, which is not. (e) Typo "picutre" removed. Also recorded the real enforcement mechanism: tests/test_service_boundaries.py, and errors.py as the neutral shared-type home.
 #### [34] Line-number citation removed from the boundary test
 *risk* - **noted**
 **test_service_boundaries.py cited the rule by line number**
 The test docstring pinned .github/instructions/services.instructions.md:13. Rewriting the file invalidated that anchor. Replaced with a section-name citation ("Structure") so future edits to the instruction file cannot silently desynchronise the test docstring. errors.py was also added to the docstring list of neutral modules.
 ### Phase 4 - measurement window
 #### [35] Both cited offenders were already deleted
 *deviation* - **noted**
 **Phase 4 premise partly overtaken by Phase 1**
 The plan states the session.commit() at line 228 "remains inside" the measurement window. Diffed against f86c0ff~1: at V4.6 the window held resolve_provider_input (async; normalization + artifact write + DB work) and that commit. Phase 1 deleted both. What remains between the clock and the wait_for is build_provider_input, now pure field copying because normalization moved to ingest and file_hash is already stored. Measured at 6.2 us per call with zero awaits, so it cannot yield to the event loop. Plan tasks 1-2 were therefore already satisfied in substance; the clock was still moved to make the property structural rather than incidental.
 #### [36] No preprocessing left to record separately
 *deviation* - **noted**
 **Plan task 3 declined**
 Task 3 offered recording preprocessing time as its own value. After Phase 1 there is no preprocessing in the window: 6.2 us of attribute copying. Adding a preprocessing_ms column to measure that is unnecessary complexity and was declined under the guiding principle. Raised rather than decided silently.
 #### [37] Undocumented 475ms contributor the plan did not identify
 *risk* - **answered**
 **Lazy provider construction was inside the timed region**
 The regression test measured 890ms where ~200ms was expected. Cause: services.sources.provider is a lazy property, and it appears as an argument expression to _call_transcriber, so it is evaluated after the clock starts but before wait_for begins timing. Measured 475ms to construct OpenRouterTranscriptionProvider on first access and 0.001ms after. The first attempt of every worker process therefore booked ~0.5s of HTTP client construction as provider latency. This plausibly accounts for the low end of the historical 0.4-2.0s local_timeout overshoot, and Phase 1 did not touch it. The property is loop-invariant, so it was hoisted above the per-source loop, which also removes the repeated attribute lookup from the two evidence-capture sites.
 #### [38] test_timeout_duration_excludes_pre_call_setup
 *comment* - **noted**
 **Regression guard added**
 New test in tests/services/test_workflows_reliability.py simulates 400ms of blocking setup against a 200ms provider budget and asserts the recorded duration_ms sits near the budget and well clear of budget+setup. Verified to fail on the pre-fix code (625 < 540 assertion error) and pass after, so it is a real guard rather than a tautology. This is the plan Phase 4 verification criterion expressed as a test.
 #### [39] sources_page.py no longer prints raw milliseconds
 *comment* - **noted**
 **Duration render scaled**
 Plan task 4. _format_duration renders >=1s as "27.6 s" and below that as "612 ms", per user selection. No test asserted the old format.
 ### Phase 5 - worker fault containment
 #### [40] Probed behaviour: the defect is a stranded job, not a silent retry
 *deviation* - **noted**
 **Plan task 4 describes a failure mode that does not occur**
 The plan asks for a test that a deliberate programming error "does not silently retry". Probed empirically with an injected AttributeError. Mode A, error raised after the claim commits (inside advance_job): raised exactly ONCE, job left at PROCESSING, retry_count 0, and never re-claimed because claim_next_queued_job filters status == QUEUED. That is a permanently stranded job with one swallowed log line, not a retry. advance_job PROCESSING branch, commented "Recover mid-flight jobs", is unreachable from the worker for the same reason. Mode B, error raised before or during the claim: 20 raises in 1.2s, an unbounded hot spin at the poll interval. The plan context says worst-case silent burn is 60s under WORKER_MAX_RETRIES=1, but Mode B never reaches the per-job retry machinery so nothing caps it. Both modes share the root cause the plan correctly identifies.
 #### [41] Flag set in 9 places, read in none
 *comment* - **noted**
 **retriable was decorative**
 Measured across src/: retriable is assigned at errors.py:40/47/79, sources.py:877/884, store.py:127/205/366, workflows.py:284/580/593/606 and read nowhere. classify_unexpected_error already returns retriable=False, so the classification existed and was discarded. Phase 5 makes it load-bearing in two places.
 #### [42] User chose: stop the worker loop
 *question* - **answered**
 **Loop policy for a non-retriable error with no job to mark**
 Mode B has no claimed job, so there is no row to mark FAILED and no reason to expect the next poll to differ. Options offered were stop the loop, circuit-breaker after N consecutive failures, or exponential backoff. User selected stopping the loop, logged at CRITICAL, returning cleanly so the exception does not surface only at app shutdown via worker_consumer_lifespan wait_for.
 #### [43] User chose: mark FAILED and keep going
 *question* - **answered**
 **Loop policy for a non-retriable error where the job CAN be marked failed**
 Distinct from entry 42 and not covered by it. Mode A can contain the failure on the job row, so stopping the loop would let one poison job halt transcription for every other job. User selected containment: mark the job FAILED, which is visible in the UI and resubmittable, and continue polling.
 #### [44] Containment write uses its own transaction
 *risk* - **noted**
 **Terminal write runs on a possibly dirty session**
 _advance_job_with_containment rolls back the caller session before marking the job FAILED, and calls update_job_state with no session so the service owns and commits its own transaction. This satisfies plan task 3 atomicity: the terminal write cannot be left half-applied by whatever failure poisoned the caller session.
 #### [45] test_run_worker_loop_survives_process_next_exception replaced
 *deviation* - **noted**
 **An existing test encoded the defective behaviour**
 That test asserted the loop SURVIVES a RuntimeError and continues, which is exactly the Mode B defect. It was replaced by test_run_worker_loop_stops_on_non_retriable_exception, plus a new test_run_worker_loop_survives_retriable_exception so suppression of genuinely transient faults stays covered. Unlike Phase 3, changing test logic here is the point of the phase. Both new guards plus the Mode A guard were verified to FAIL on pre-fix code: the Mode B test times out, which is the infinite spin made visible.
 ### Phase 6 - CI enforcement
 #### [46] Remote is Gitea 1.27.2, not GitHub
 *comment* - **noted**
 **The plan assumes GitHub Actions**
 Remote is bbchops/transcription on Gitea 1.27.2, which reads .github/workflows/ and proxies actions/checkout@v4 to GitHub. Workflow syntax needed no change. Note the remote default branch is traumatized, not main.
 #### [47] Runner availability cannot be confirmed via the API
 *risk* - **noted**
 **Repo-scoped runner list returns 0; admin endpoint returns 403**
 Existence of CI could not be asserted by query on this host. Proven instead by observation: an instance-level runner named docker-runner executed the jobs. Anyone re-verifying this must trigger a run rather than trust the runner API.
 #### [48] CI writes a .env file instead of exporting an env var
 *deviation* - **noted**
 **Settings reads the .env file; the external-test skip guard reads os.getenv**
 The two read different sources, and locally both conditions hold at once, which is why 4 tests skip. Measured in CI: no .env = 115 failed / 18 errors; exported dummy var = 3 failed (externals un-skip and hit the network); written .env file = the exact local baseline. Only openrouter_api_key is required.
 #### [49] CI invokes pre-commit rather than repeating ruff/ty commands
 *deviation* - **noted**
 **Plan task 2 asks CI to run the same checks as local**
 Satisfied structurally rather than by copying command strings: CI runs uv run pre-commit run --all-files, so the checks have a single definition in .pre-commit-config.yaml and CI cannot drift from local. Hooks are language: system and uv run puts .venv on PATH.
 #### [50] Platform-dependent prompt name guard, caught by CI on its first green run
 *deviation* - **noted**
 **The direct-child name guard relied on Path(name).name != name**
 On POSIX, backslash is an ordinary filename character, so nested\prompt.md passed the direct-child guard and failed later as NOT_FOUND instead of VALIDATION. Windows can never reproduce it. No traversal was possible because the path.parent != root check still held, so severity is a wrong error category plus a red gate. Fixed by rejecting / and \ explicitly, matching the ^[^/\\]+$ pattern config.PromptFilename already used. User approved the code fix over weakening the test.
@@ -9,12 +9,21 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 from ..config import Settings
 from .documents import DocumentService
 from .evidence import EvidenceService
 from .jobs import JobService
 from .people import PeopleService
 from .prompts import PromptStore
 from .sources import SourceService
-__all__ = ["DocumentService", "JobService", "PeopleService", "PromptStore", "ServiceBundle", "SourceService"]
+__all__ = [
    "DocumentService",
    "EvidenceService",
    "JobService",
    "PeopleService",
    "PromptStore",
    "ServiceBundle",
    "SourceService",
 ]
@dataclass(frozen=True, slots=True)
@@ -25,6 +34,7 @@ class ServiceBundle:
    sources: SourceService = field(default_factory=SourceService)
    jobs: JobService = field(default_factory=JobService)
    people: PeopleService = field(default_factory=PeopleService)
    evidence: EvidenceService = field(default_factory=EvidenceService)
    @classmethod
    def from_session_factory(
@@ -41,6 +51,7 @@ class ServiceBundle:
            sources=SourceService(session_factory=session_factory, settings=settings),
            jobs=JobService(session_factory=session_factory, settings=settings),
            people=PeopleService(session_factory=session_factory, settings=settings),
            evidence=EvidenceService(session_factory=session_factory, settings=settings),
        )
    async def aclose(self) -> None:
@@ -0,0 +1,32 @@
 """Error vocabulary shared across the source, evidence, and prompt services.
 These live in a neutral module rather than in the service that raises them
 because more than one service raises them, and ``services.instructions.md``
 forbids a service module from importing a sibling. Orchestration modules and
 the UI import from here, so the exception a caller catches does not change when
 an operation moves between services.
 """
 from __future__ import annotations
 from transcription.errors import AppError
 class PromptLoadError(AppError):
    """Raised when prompt artifacts cannot be loaded safely."""
 class TranscriptionError(AppError):
    """Raised when transcription execution fails."""
 class TranscriptionNotFoundError(TranscriptionError):
    """Raised when a transcription-related resource is not found."""
 class SourceDeleteBlockedError(TranscriptionError):
    """Raised when source deletion is blocked by dependency policy."""
 class CandidatePromotionError(TranscriptionError):
    """Raised when a machine attempt cannot be selected for its Source."""
@@ -0,0 +1,210 @@
 """Read and export the immutable execution evidence trail.
 ``ExecutionAttempt`` is append-only: one row per provider call, written once by
 the transcription workflow and never updated. Everything here is therefore a
 read, a projection, or an export, with one exception - ``promote_machine_attempt``
 selects which attempt a ``Source`` presents, which is an evidence decision even
 though the write lands on ``Source``.
 """
 from __future__ import annotations
 import base64
 import hashlib
 from collections.abc import Sequence
 from dataclasses import dataclass
 from uuid import UUID
 from pydantic import JsonValue
 from sqlalchemy import inspect as sqlalchemy_inspect
 from sqlmodel import col
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 from transcription.db.models import ExecutionAttempt
 from transcription.db.models import JobSourceStatus
 from transcription.db.models import Source
 from transcription.errors import ErrorCategory
 from ..db.loading import defer
 from .base import ServiceBase
 from .errors import CandidatePromotionError
 from .errors import TranscriptionNotFoundError
@dataclass(frozen=True, slots=True)
 class LatestExecutionAttempt:
    """One execution attempt plus the loader facts a caller needs to render it."""
    attempt: ExecutionAttempt
    transport_body_deferred: bool
 class EvidenceService(ServiceBase):
    """Read, project, and export execution attempt evidence."""
    async def read_latest_execution_attempt(
        self,
        *,
        job_source_id: UUID,
        session: AsyncSession | None = None,
    ) -> LatestExecutionAttempt | None:
        """Read only the latest immutable attempt for one compatibility projection.
        The transport body is deferred because it can be arbitrarily large; the
        returned read model reports that as a plain flag so callers never have to
        inspect ORM loader state.
        """
        async with self._session_scope(session) as _session:
            query = (
                select(ExecutionAttempt)
                .options(defer(ExecutionAttempt.transport_body))
                .where(ExecutionAttempt.job_source_id == job_source_id)
                .order_by(
                    col(ExecutionAttempt.attempt_number).desc(),
                    col(ExecutionAttempt.id).desc(),
                )
                .limit(1)
            )
            attempt = (await _session.exec(query)).first()
            if attempt is None:
                return None
            deferred = "transport_body" in sqlalchemy_inspect(attempt).unloaded
            return LatestExecutionAttempt(attempt=attempt, transport_body_deferred=deferred)
    async def list_execution_attempts(
        self,
        *,
        source_id: UUID | None = None,
        job_id: UUID | None = None,
        session: AsyncSession | None = None,
    ) -> Sequence[ExecutionAttempt]:
        """List immutable execution evidence in stable attempt order."""
        async with self._session_scope(session) as _session:
            query = select(ExecutionAttempt)
            if source_id is not None:
                query = query.where(ExecutionAttempt.source_id == source_id)
            if job_id is not None:
                query = query.where(ExecutionAttempt.job_id == job_id)
            query = query.order_by(
                col(ExecutionAttempt.job_id),
                col(ExecutionAttempt.source_id),
                col(ExecutionAttempt.attempt_number),
                col(ExecutionAttempt.id),
            )
            return (await _session.exec(query)).all()
    async def promote_machine_attempt(
        self,
        *,
        source_id: UUID,
        execution_attempt_id: UUID,
        session: AsyncSession | None = None,
    ) -> Source:
        """Atomically select one successful machine attempt as the Source projection."""
        async with self._session_scope(session) as _session:
            source = await self._read_source(
                session=_session,
                source_id=source_id,
                suggestion="Refresh Source Detail and retry.",
            )
            attempt = await _session.get(ExecutionAttempt, execution_attempt_id)
            if (
                attempt is None
                or attempt.source_id != source_id
                or attempt.status != JobSourceStatus.TRANSCRIBED
                or not attempt.raw_transcription
            ):
                raise CandidatePromotionError(
                    "Only a successful transcription attempt belonging to this Source can be selected",
                    category=ErrorCategory.VALIDATION,
                    suggestion="Select an available successful candidate from Source Detail.",
                )
            source.preferred_execution_attempt_id = attempt.id
            source.raw_transcription = attempt.raw_transcription
            await self._finalize(session=_session, caller_session=session, refresh=(source,))
            return source
    async def build_evidence_export(
        self,
        *,
        source_id: UUID,
        session: AsyncSession | None = None,
    ) -> dict[str, JsonValue]:
        """Build a versioned, source-reference-only evidence export."""
        async with self._session_scope(session) as _session:
            source = await self._read_source(session=_session, source_id=source_id)
            attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session))
        attempt_payloads = [
            {
                "id": str(attempt.id),
                "job_id": str(attempt.job_id),
                "source_id": str(attempt.source_id),
                "attempt_number": attempt.attempt_number,
                "status": attempt.status.value,
                "provider": attempt.provider,
                "model": attempt.model,
                "request_manifest": attempt.request_manifest,
                "request_manifest_sha256": attempt.request_manifest_sha256,
                "request_manifest_schema_version": attempt.request_manifest_schema_version,
                "transport": {
                    "response_received": attempt.response_received,
                    "status_code": attempt.transport_status_code,
                    "body_base64": (
                        base64.b64encode(attempt.transport_body).decode("ascii")
                        if attempt.transport_body is not None
                        else None
                    ),
                    "body_sha256": (
                        hashlib.sha256(attempt.transport_body).hexdigest()
                        if attempt.transport_body is not None
                        else None
                    ),
                    "content_type": attempt.transport_content_type,
                    "content_encoding": attempt.transport_content_encoding,
                    "safe_headers": attempt.transport_safe_headers,
                    "request_id": attempt.router_request_id,
                    "generation_id": attempt.router_generation_id,
                },
                "sdk_response_snapshot": attempt.sdk_response_snapshot,
                "normalized_metadata": attempt.normalized_metadata,
                "software_context": attempt.software_context,
                "raw_transcription": attempt.raw_transcription,
                "error_category": attempt.error_category,
                "error_detail": attempt.error_detail,
                "failure_phase": attempt.failure_phase,
                "started_at": attempt.started_at.isoformat(),
                "finished_at": attempt.finished_at.isoformat(),
                "duration_ms": attempt.duration_ms,
            }
            for attempt in attempts
        ]
        return {
            "schema_name": "transcription.evidence-export",
            "schema_version": "1",
            "source": {
                "id": str(source.id),
                "digest_sha256": source.file_hash,
                "byte_size": source.file_size_bytes,
                "page_number": source.page_number,
                "upload_name": source.upload_name,
            },
            "attempts": attempt_payloads,
        }
    async def _read_source(
        self,
        *,
        session: AsyncSession,
        source_id: UUID,
        suggestion: str = "Verify the source id and retry.",
    ) -> Source:
        return await self._get_or_raise(
            Source,
            source_id,
            session=session,
            error=TranscriptionNotFoundError,
            noun="Source",
            suggestion=suggestion,
        )
@@ -104,8 +104,11 @@ class PromptStore:
    def _resolve_existing_prompt(self, name: str) -> Path:
        normalized_name = name.strip()
        # Path().name is platform-dependent: POSIX treats "\" as an ordinary filename
        # character, so reject both separators explicitly to match config.PromptFilename.
        if (
            not normalized_name
            or any(separator in normalized_name for separator in ("/", "\\"))
            or Path(normalized_name).name != normalized_name
            or Path(normalized_name).suffix.lower() != PROMPT_EXTENSION
        ):
@@ -3,7 +3,6 @@
 from __future__ import annotations
 import asyncio
 import base64
 import hashlib
 import logging
 from collections.abc import Sequence
@@ -22,7 +21,6 @@ from pydantic import JsonValue
 from pydantic import TypeAdapter
 from pydantic import ValidationError
 from sqlalchemy import func
 from sqlalchemy import inspect as sqlalchemy_inspect
 from sqlalchemy import literal
 from sqlalchemy import tuple_
 from sqlalchemy.ext.asyncio import async_sessionmaker
@@ -37,7 +35,6 @@ from transcription.db.models import Job
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
 from transcription.db.models import Source
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
@@ -50,10 +47,13 @@ from transcription.providers import TranscriptionResult
 from transcription.providers import TransportEvidence
 from transcription.providers import get_transcription_provider
 from ..db.loading import defer
 from ..db.loading import orm_attribute
 from ..db.loading import selectinload
 from .base import ServiceBase
 from .errors import PromptLoadError
 from .errors import SourceDeleteBlockedError
 from .errors import TranscriptionError
 from .errors import TranscriptionNotFoundError
 from .source_media import lookup_source_mime_type
 from .source_media import supported_source_formats
@@ -76,26 +76,6 @@ class PromptExecution(BaseModel):
    top_p: float | None = Field(ge=0.0, le=1.0)
 class PromptLoadError(AppError):
    """Raised when prompt artifacts cannot be loaded safely."""
 class TranscriptionError(AppError):
    """Raised when transcription execution fails."""
 class TranscriptionNotFoundError(TranscriptionError):
    """Raised when a transcription-related resource is not found."""
 class SourceDeleteBlockedError(TranscriptionError):
    """Raised when source deletion is blocked by dependency policy."""
 class CandidatePromotionError(TranscriptionError):
    """Raised when a machine attempt cannot be selected for its Source."""
@dataclass(frozen=True, slots=True)
 class SourceNavigation:
    """Adjacent Source identifiers within one ordered Document."""
@@ -128,14 +108,6 @@ def build_provider_input(source: Source) -> ProviderInput:
    )
@dataclass(frozen=True, slots=True)
 class LatestExecutionAttempt:
    """One execution attempt plus the loader facts a caller needs to render it."""
    attempt: ExecutionAttempt
    transport_body_deferred: bool
 class SourceService(ServiceBase):
    """Manage source records, media payloads, revisions, and page execution output."""
@@ -214,35 +186,6 @@ class SourceService(ServiceBase):
            )
        return source
    async def read_latest_execution_attempt(
        self,
        *,
        job_source_id: UUID,
        session: AsyncSession | None = None,
    ) -> LatestExecutionAttempt | None:
        """Read only the latest immutable attempt for one compatibility projection.
        The transport body is deferred because it can be arbitrarily large; the
        returned read model reports that as a plain flag so callers never have to
        inspect ORM loader state.
        """
        async with self._session_scope(session) as _session:
            query = (
                select(ExecutionAttempt)
                .options(defer(ExecutionAttempt.transport_body))
                .where(ExecutionAttempt.job_source_id == job_source_id)
                .order_by(
                    col(ExecutionAttempt.attempt_number).desc(),
                    col(ExecutionAttempt.id).desc(),
                )
                .limit(1)
            )
            attempt = (await _session.exec(query)).first()
            if attempt is None:
                return None
            deferred = "transport_body" in sqlalchemy_inspect(attempt).unloaded
            return LatestExecutionAttempt(attempt=attempt, transport_body_deferred=deferred)
    async def read_source_navigation(
        self,
        source_id: UUID,
@@ -640,127 +583,6 @@ class SourceService(ServiceBase):
            await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source, attempt))
            return job_source
    async def promote_machine_attempt(
        self,
        *,
        source_id: UUID,
        execution_attempt_id: UUID,
        session: AsyncSession | None = None,
    ) -> Source:
        """Atomically select one successful machine attempt as the Source projection."""
        async with self._session_scope(session) as _session:
            source = await self._read_source(
                session=_session,
                source_id=source_id,
                suggestion="Refresh Source Detail and retry.",
            )
            attempt = await _session.get(ExecutionAttempt, execution_attempt_id)
            if (
                attempt is None
                or attempt.source_id != source_id
                or attempt.status != JobSourceStatus.TRANSCRIBED
                or not attempt.raw_transcription
            ):
                raise CandidatePromotionError(
                    "Only a successful transcription attempt belonging to this Source can be selected",
                    category=ErrorCategory.VALIDATION,
                    suggestion="Select an available successful candidate from Source Detail.",
                )
            source.preferred_execution_attempt_id = attempt.id
            source.raw_transcription = attempt.raw_transcription
            await self._finalize(session=_session, caller_session=session, refresh=(source,))
            return source
    async def list_execution_attempts(
        self,
        *,
        source_id: UUID | None = None,
        job_id: UUID | None = None,
        session: AsyncSession | None = None,
    ) -> Sequence[ExecutionAttempt]:
        """List immutable execution evidence in stable attempt order."""
        async with self._session_scope(session) as _session:
            query = select(ExecutionAttempt)
            if source_id is not None:
                query = query.where(ExecutionAttempt.source_id == source_id)
            if job_id is not None:
                query = query.where(ExecutionAttempt.job_id == job_id)
            query = query.order_by(
                col(ExecutionAttempt.job_id),
                col(ExecutionAttempt.source_id),
                col(ExecutionAttempt.attempt_number),
                col(ExecutionAttempt.id),
            )
            return (await _session.exec(query)).all()
    async def build_evidence_export(
        self,
        *,
        source_id: UUID,
        session: AsyncSession | None = None,
    ) -> dict[str, JsonValue]:
        """Build a versioned, source-reference-only evidence export."""
        async with self._session_scope(session) as _session:
            source = await self._read_source(session=_session, source_id=source_id)
            attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session))
        attempt_payloads = [
            {
                "id": str(attempt.id),
                "job_id": str(attempt.job_id),
                "source_id": str(attempt.source_id),
                "attempt_number": attempt.attempt_number,
                "status": attempt.status.value,
                "provider": attempt.provider,
                "model": attempt.model,
                "request_manifest": attempt.request_manifest,
                "request_manifest_sha256": attempt.request_manifest_sha256,
                "request_manifest_schema_version": attempt.request_manifest_schema_version,
                "transport": {
                    "response_received": attempt.response_received,
                    "status_code": attempt.transport_status_code,
                    "body_base64": (
                        base64.b64encode(attempt.transport_body).decode("ascii")
                        if attempt.transport_body is not None
                        else None
                    ),
                    "body_sha256": (
                        hashlib.sha256(attempt.transport_body).hexdigest()
                        if attempt.transport_body is not None
                        else None
                    ),
                    "content_type": attempt.transport_content_type,
                    "content_encoding": attempt.transport_content_encoding,
                    "safe_headers": attempt.transport_safe_headers,
                    "request_id": attempt.router_request_id,
                    "generation_id": attempt.router_generation_id,
                },
                "sdk_response_snapshot": attempt.sdk_response_snapshot,
                "normalized_metadata": attempt.normalized_metadata,
                "software_context": attempt.software_context,
                "raw_transcription": attempt.raw_transcription,
                "error_category": attempt.error_category,
                "error_detail": attempt.error_detail,
                "failure_phase": attempt.failure_phase,
                "started_at": attempt.started_at.isoformat(),
                "finished_at": attempt.finished_at.isoformat(),
                "duration_ms": attempt.duration_ms,
            }
            for attempt in attempts
        ]
        return {
            "schema_name": "transcription.evidence-export",
            "schema_version": "1",
            "source": {
                "id": str(source.id),
                "digest_sha256": source.file_hash,
                "byte_size": source.file_size_bytes,
                "page_number": source.page_number,
                "upload_name": source.upload_name,
            },
            "attempts": attempt_payloads,
        }
    async def upsert_revision_for_source(
        self,
        *,
@@ -23,10 +23,10 @@ from ..db.models import JobSourceStatus
 from ..db.models import Source
 from ..db.session import SessionFactory
 from ..db.session import session_scope
 from .errors import TranscriptionError
 from .media_storage import build_stored_filename
 from .media_storage import write_media_bytes
 from .normalization import normalize_orientation_async
 from .sources import TranscriptionError
 from .sources import build_prompt_execution
 from .sources import source_mime_type
 from .sources import validate_source_content
@@ -205,6 +205,10 @@ async def process_queued_job(  # noqa: PLR0915
    externally_stopped = False
    prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings)
    # Resolve the lazy provider property once, outside the timed region. First access
    # constructs the HTTP client (~0.5s), which would otherwise be booked as provider
    # latency on the first attempt of every worker process (review log [55]).
    provider = services.sources.provider
    for source in sources:
        if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
@@ -212,6 +216,8 @@ async def process_queued_job(  # noqa: PLR0915
            break
        started_at = datetime.now(UTC)
        # Fallback start for failures raised before the provider call; reset to the
        # true call boundary immediately before the wait_for below.
        monotonic_started_at = asyncio.get_running_loop().time()
        result: TranscriptionResult | None = None
        provider_input = None
@@ -225,12 +231,16 @@ async def process_queued_job(  # noqa: PLR0915
                media_type=provider_input.media_type,
                page_number=source.page_number,
            )
            # Restart the clock so duration_ms covers only what the wait_for below
            # governs. The pre-loop assignment stays as the fallback for failures
            # raised before this point, which would otherwise leave it unbound.
            monotonic_started_at = asyncio.get_running_loop().time()
            result = await asyncio.wait_for(
                _call_transcriber(
                    input_path=provider_input.path,
                    prompt_execution=prompt_execution,
                    settings=runtime_settings,
-                    provider=services.sources.provider,
+                    provider=provider,
                    source_reference=source_reference,
                    requested_model=source_job.model,
                ),
@@ -283,8 +293,8 @@ async def process_queued_job(  # noqa: PLR0915
                    0,
                    int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
                ),
-                request_manifest=services.sources.provider.current_request_manifest,
+                request_manifest=provider.current_request_manifest,
-                transport_evidence=services.sources.provider.current_transport_evidence,
+                transport_evidence=provider.current_transport_evidence,
                failure_phase="local_timeout",
            )
            failed_pages.append(page_outcome)
@@ -406,10 +416,46 @@ async def process_next_queued_job(
    if session is not None:
        await session.commit()
-    await advance_job(job=job, services=services, settings=settings, session=session)
+    await _advance_job_with_containment(job=job, services=services, settings=settings, session=session)
    return True
 async def _advance_job_with_containment(
    *,
    job: Job,
    services: ServiceBundle,
    settings: Settings | None,
    session: AsyncSession | None,
 ) -> None:
    """Advance a claimed job, guaranteeing it never stays stuck in PROCESSING.
    The job has already been committed as PROCESSING at this point, and
    ``claim_next_queued_job`` only ever selects QUEUED rows. So an exception
    escaping ``advance_job`` used to strand the job in PROCESSING permanently,
    with a single swallowed log line and no recovery path (review log [8]).
    Any escaping exception is therefore classified and the job driven to the
    terminal FAILED state, which is visible in the UI and resubmittable. The
    terminal write runs in its own transaction so it is atomic even when the
    caller's session was left dirty by the failure.
    """
    try:
        await advance_job(job=job, services=services, settings=settings, session=session)
    except Exception as exc:
        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.advance_job")
        logger.exception(
            "Job processing failed outside page handling operation=worker.advance_job "
            "job_id=%s error_id=%s category=%s retriable=%s",
            job.id,
            error.error_id,
            error.category.value,
            error.retriable,
        )
        if session is not None:
            await session.rollback()
        await services.jobs.update_job_state(job_id=job.id, status=JobStatus.FAILED)
 def _resolve_job_sources(job: Job) -> list[Source]:
    """Resolve pending linked sources for a job in deterministic page order.
@@ -14,10 +14,11 @@ from transcription.db.models import ExecutionAttempt
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
 from transcription.db.models import Source
-from transcription.services.sources import LatestExecutionAttempt
+from transcription.services.errors import SourceDeleteBlockedError
-from transcription.services.sources import SourceDeleteBlockedError
+from transcription.services.errors import TranscriptionNotFoundError
 from transcription.services.evidence import EvidenceService
 from transcription.services.evidence import LatestExecutionAttempt
 from transcription.services.sources import SourceService
 from transcription.services.sources import TranscriptionNotFoundError
 from transcription.ui.components.app_shell import render_navigation_header
 from transcription.ui.components.cards import archival_card
 from transcription.ui.components.confirm_delete import render_delete_actions
@@ -112,6 +113,7 @@ def register_page() -> None:  # noqa: PLR0915
    @ui.page("/sources/{source_id}")
    async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
        sources_service = SourceService(session_factory=session_factory)
        evidence_service = EvidenceService(session_factory=session_factory)
        render_navigation_header(current_path="/sources")
        parsed_source_id = parsed_record_id(source_id, noun="Source")
@@ -123,11 +125,11 @@ def register_page() -> None:  # noqa: PLR0915
            navigation = await sources_service.read_source_navigation(parsed_source_id)
            latest_job_source = source.latest_job_source
            latest_attempt = (
-                await sources_service.read_latest_execution_attempt(job_source_id=latest_job_source.id)
+                await evidence_service.read_latest_execution_attempt(job_source_id=latest_job_source.id)
                if latest_job_source is not None
                else None
            )
-            attempts = list(await sources_service.list_execution_attempts(source_id=parsed_source_id))
+            attempts = list(await evidence_service.list_execution_attempts(source_id=parsed_source_id))
        except TranscriptionNotFoundError:
            render_record_not_found("Source")
            return
@@ -158,7 +160,7 @@ def register_page() -> None:  # noqa: PLR0915
                        "Export Evidence",
                        on_click=lambda: _download_evidence(
                            source_id=source.id,
-                            sources_service=sources_service,
+                            evidence_service=evidence_service,
                        ),
                        icon="download",
                    ).props("flat")
@@ -185,7 +187,7 @@ def register_page() -> None:  # noqa: PLR0915
                _render_machine_candidates(
                    source=source,
                    attempts=attempts,
-                    sources_service=sources_service,
+                    evidence_service=evidence_service,
                )
                _render_source_metadata_column(
                    source=source,
@@ -364,6 +366,13 @@ def _render_source_job_metadata_zone(
        _render_provider_evidence(latest_attempt=latest_attempt)
 def _format_duration(duration_ms: int) -> str:
    """Render an attempt duration with a unit that suits its magnitude."""
    if duration_ms >= 1000:
        return f"{duration_ms / 1000:.1f} s"
    return f"{duration_ms} ms"
 def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None:
    ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3")
    if latest_attempt is None:
@@ -372,7 +381,7 @@ def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None)
    attempt = latest_attempt.attempt
    metadata_row("Attempt:", str(attempt.attempt_number))
-    metadata_row("Duration:", f"{attempt.duration_ms} ms")
+    metadata_row("Duration:", _format_duration(attempt.duration_ms))
    _render_json_evidence("Request Manifest", attempt.request_manifest)
    _render_json_evidence("Transport Response", _transport_display(latest_attempt))
    _render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot)
@@ -409,9 +418,9 @@ def _transport_display(latest_attempt: LatestExecutionAttempt) -> dict[str, obje
    }
-async def _download_evidence(*, source_id: UUID, sources_service: SourceService) -> None:
+async def _download_evidence(*, source_id: UUID, evidence_service: EvidenceService) -> None:
    try:
-        payload = await sources_service.build_evidence_export(source_id=source_id)
+        payload = await evidence_service.build_evidence_export(source_id=source_id)
    except Exception as exc:  # noqa: BLE001
        show_error(exc, title="Export failed", operation="sources.evidence_export")
        return
@@ -521,7 +530,7 @@ def _render_machine_candidates(
    *,
    source: Source,
    attempts: list[ExecutionAttempt],
-    sources_service: SourceService,
+    evidence_service: EvidenceService,
 ) -> None:
    successful = [
        attempt
@@ -581,7 +590,7 @@ def _render_machine_candidates(
                    async def promote(candidate_id: UUID = attempt.id) -> None:
                        try:
-                            await sources_service.promote_machine_attempt(
+                            await evidence_service.promote_machine_attempt(
                                source_id=source.id,
                                execution_attempt_id=candidate_id,
                            )
@@ -94,16 +94,30 @@ async def worker_consumer_lifespan(
@contextmanager
 def handle_worker_exceptions(operation: str = "worker.loop"):
-    """Context manager to log and suppress exceptions in the worker loop."""
+    """Log worker-loop exceptions, suppressing only the retriable ones.
    ``classify_unexpected_error`` marks unknown exceptions non-retriable, but that
    verdict used to be logged and then discarded, so a programming error raised
    before a job could be claimed spun the loop at the poll interval indefinitely.
    Nothing capped it, because it never reached the per-job retry machinery
    (review log [8]).
    A non-retriable fault is a defect rather than a transient condition, so it is
    re-raised for the caller to stop on. Faults raised after a job is claimed are
    contained at the job level instead, and never reach here.
    """
    try:
        yield
    except Exception as exc:
        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation=operation)
        logger.exception(
-            "Worker loop exception error_id=%s category=%s",
+            "Worker loop exception error_id=%s category=%s retriable=%s",
            error.error_id,
            error.category.value,
            error.retriable,
        )
        if not error.retriable:
            raise error from exc
 async def run_worker_loop(
@@ -121,6 +135,10 @@ async def run_worker_loop(
    The service bundle — and with it the provider's pooled HTTP client — is built
    once for the lifetime of the loop, so consecutive jobs reuse one connection
    instead of paying a fresh TLS handshake each time.
    The loop returns early on a non-retriable error raised before a job could be
    claimed. Faults raised after a claim are contained by marking that job FAILED,
    so a single poison job cannot stop transcription for every other job.
    """
    services = ServiceBundle.from_session_factory(session_factory)
    try:
@@ -150,6 +168,15 @@ async def run_worker_loop(
            if wake_event is None and not processed_any:
                await asyncio.sleep(poll_interval_seconds)
    except AppError as error:
        # Stop rather than spin. A non-retriable fault here means no job could be
        # claimed, so there is no row to mark FAILED and no reason to expect the
        # next poll to behave differently.
        logger.critical(
            "Worker loop stopped after a non-retriable error error_id=%s category=%s",
            error.error_id,
            error.category.value,
        )
    finally:
        await services.aclose()
@@ -13,6 +13,7 @@ from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.db.models import Source
 from transcription.services.documents import DocumentService
 from transcription.services.evidence import EvidenceService
 from transcription.services.jobs import JobCancelBlockedError
 from transcription.services.jobs import JobDeleteBlockedError
 from transcription.services.jobs import JobNotFoundError
@@ -274,6 +275,7 @@ class TestJobService:
        document_service: DocumentService,
    ):
        source_service = SourceService(session_factory=job_service.session_factory)
        evidence_service = EvidenceService(session_factory=job_service.session_factory)
        document = await document_service.create_document(Document(name="evidence-delete-doc"))
        job = await job_service.create_job(Job(document_id=document.id, status=JobStatus.FAILED))
        source = await source_service.create_source(
@@ -299,7 +301,7 @@ class TestJobService:
        with pytest.raises(JobNotFoundError):
            await job_service.read_job(job_id=job.id)
-        assert await source_service.list_execution_attempts(job_id=job.id) == []
+        assert await evidence_service.list_execution_attempts(job_id=job.id) == []
        assert (await source_service.read_source(source.id)).id == source.id
    @pytest.mark.asyncio
@@ -13,10 +13,10 @@ from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.db.models import Source
 from transcription.services.documents import DocumentService
 from transcription.services.errors import SourceDeleteBlockedError
 from transcription.services.errors import TranscriptionNotFoundError
 from transcription.services.jobs import JobService
 from transcription.services.sources import SourceDeleteBlockedError
 from transcription.services.sources import SourceService
 from transcription.services.sources import TranscriptionNotFoundError
@pytest.mark.integration
@@ -13,10 +13,11 @@ from transcription.db.models import Source
 from transcription.errors import ErrorCategory
 from transcription.services.documents import DocumentDeleteBlockedError
 from transcription.services.documents import DocumentService
 from transcription.services.errors import SourceDeleteBlockedError
 from transcription.services.evidence import EvidenceService
 from transcription.services.jobs import JobService
 from transcription.services.people import PeopleError
 from transcription.services.people import PeopleService
 from transcription.services.sources import SourceDeleteBlockedError
 from transcription.services.sources import SourceService
@@ -324,7 +325,8 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
    assert len(stored_rows) == 1
    assert stored_rows[0].status == JobSourceStatus.TRANSCRIBED
-    attempt = await transcriptions.read_latest_execution_attempt(job_source_id=stored_rows[0].id)
+    evidence = EvidenceService(session_factory=transcriptions.session_factory)
    attempt = await evidence.read_latest_execution_attempt(job_source_id=stored_rows[0].id)
    assert attempt is not None
    assert attempt.attempt.raw_transcription == "provider transcript"
    assert attempt.attempt.normalized_metadata == metadata
@@ -11,19 +11,12 @@ from transcription.db.models import JobPurpose
 from transcription.db.models import JobSource
 from transcription.db.models import Source
 from transcription.services import ServiceBundle
-from transcription.services.documents import DocumentService
+from transcription.services.errors import CandidatePromotionError
 from transcription.services.jobs import JobService
 from transcription.services.sources import CandidatePromotionError
 from transcription.services.sources import SourceService
 from transcription.services.workflows import create_source_retranscription_job
 def _services(default_session_factory, settings: Settings) -> ServiceBundle:
-    return ServiceBundle(
+    return ServiceBundle.from_session_factory(default_session_factory, settings=settings)
        documents=DocumentService(session_factory=default_session_factory, settings=settings),
        jobs=JobService(session_factory=default_session_factory, settings=settings),
        sources=SourceService(session_factory=default_session_factory, settings=settings),
    )
 async def _seed_source(services: ServiceBundle) -> Source:
@@ -71,14 +64,14 @@ async def test_first_success_is_preferred_and_later_success_remains_candidate(de
        model="model-b",
    )
    unchanged = await services.sources.read_source(source.id)
-    attempts = await services.sources.list_execution_attempts(source_id=source.id)
+    attempts = await services.evidence.list_execution_attempts(source_id=source.id)
    assert unchanged.raw_transcription == "first result"
    assert unchanged.preferred_execution_attempt_id == first_attempt_id
    assert {attempt.raw_transcription for attempt in attempts} == {"first result", "candidate result"}
    candidate = next(attempt for attempt in attempts if attempt.raw_transcription == "candidate result")
-    promoted = await services.sources.promote_machine_attempt(
+    promoted = await services.evidence.promote_machine_attempt(
        source_id=source.id,
        execution_attempt_id=candidate.id,
    )
@@ -95,7 +88,7 @@ async def test_promotion_rejects_unrelated_attempt(default_session_factory):
    source = await _seed_source(services)
    with pytest.raises(CandidatePromotionError):
-        await services.sources.promote_machine_attempt(
+        await services.evidence.promote_machine_attempt(
            source_id=source.id,
            execution_attempt_id=uuid4(),
        )
@@ -1,6 +1,7 @@
 """Reliability tests for worker workflow timeout behavior."""
 import asyncio
 import time
 from pathlib import Path
 from uuid import uuid4
@@ -18,6 +19,7 @@ from transcription.db.models import JobStatus
 from transcription.db.models import Source
 from transcription.providers.base import TranscriptionResult
 from transcription.services import ServiceBundle
 from transcription.services import workflows as workflows_module
 from transcription.services.workflows import process_queued_job
@@ -103,18 +105,143 @@ class TestWorkflowReliability:
        assert "timed out" in error_detail.lower()
        assert "20.0s" in error_detail
    @pytest.mark.asyncio
    async def test_timeout_duration_excludes_pre_call_setup(self, default_session_factory, monkeypatch):
        """duration_ms covers only the provider call, not the setup preceding it.
        Regression guard for review log [55]: three historical ``local_timeout`` rows
        recorded 0.4-2.0 s more than the configured budget because the measurement
        window opened before payload resolution. Blocking setup is simulated here so
        the assertion fails if that window ever reopens.
        """
        services = ServiceBundle.from_session_factory(default_session_factory)
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="window-doc")
            session.add(document)
            await session.flush()
            job = Job(document_id=document.id, status=JobStatus.QUEUED)
            session.add(job)
            await session.flush()
            source = Source(
                document_id=document.id,
                page_number=1,
                upload_name="window.jpg",
                filename="window.jpg",
                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                file_hash="d" * 64,
                file_size_bytes=1,
            )
            session.add(source)
            await session.flush()
            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
            await session.commit()
            loaded = await services.jobs.read_job(job_id=job.id, session=session)
        setup_seconds = 0.40
        budget_seconds = 0.20
        real_build = workflows_module.build_provider_input
        def _slow_build(source_arg):
            time.sleep(setup_seconds)
            return real_build(source_arg)
        async def _never_returns(*args, **kwargs):
            _ = (args, kwargs)
            await asyncio.sleep(budget_seconds * 20)
        monkeypatch.setattr("transcription.services.workflows.build_provider_input", _slow_build)
        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns)
        result = await process_queued_job(
            job=loaded,
            services=services,
            settings=Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=budget_seconds),
        )
        assert result is not None
        assert result.status == JobStatus.FAILED
        async with services.jobs._session_scope() as session:
            attempts = (
                (
                    await session.exec(
                        select(ExecutionAttempt).where(
                            col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                        )
                    )
                )
                .all()
            )
        assert len(attempts) == 1
        duration_ms = attempts[0].duration_ms
        # At or just above the budget, and well clear of budget + setup.
        assert duration_ms >= int(budget_seconds * 1000 * 0.9)
        assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9)
    @pytest.mark.asyncio
    async def test_error_after_claim_fails_the_job_instead_of_stranding_it(
        self,
        default_session_factory,
        monkeypatch,
    ):
        """A non-retriable fault after the claim drives the job terminal, not stuck.
        Regression guard for review log [8]. The claim commits PROCESSING before any
        provider work, and claim_next_queued_job only ever selects QUEUED, so an
        exception escaping advance_job used to strand the job in PROCESSING forever
        with one swallowed log line. Measured before the fix: raised once, job left
        processing, retry_count 0, never re-claimed.
        """
        services = ServiceBundle.from_session_factory(default_session_factory)
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="strand-doc")
            session.add(document)
            await session.flush()
            job = Job(document_id=document.id, status=JobStatus.QUEUED)
            session.add(job)
            await session.flush()
            source = Source(
                document_id=document.id,
                page_number=1,
                upload_name="strand.jpg",
                filename="strand.jpg",
                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                file_hash="e" * 64,
                file_size_bytes=1,
            )
            session.add(source)
            await session.flush()
            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
            await session.commit()
            job_id = job.id
        async def _succeeds(*args, **kwargs):
            _ = (args, kwargs)
            return TranscriptionResult(text="page text", provider="test", model="test-model")
        async def _boom(**kwargs):
            _ = kwargs
            raise AttributeError("deliberate programming error")
        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds)
        monkeypatch.setattr("transcription.services.workflows._finalize_batch_outcome", _boom)
        processed = await workflows_module.process_next_queued_job(services=services)
        assert processed is True
        async with services.jobs._session_scope() as session:
            final = await session.get(Job, job_id)
            assert final is not None
            # Terminal and resubmittable, rather than stranded in PROCESSING.
            assert final.status == JobStatus.FAILED
    @pytest.mark.asyncio
    async def test_completed_page_is_committed_before_next_provider_call_finishes(
        self,
        default_session_factory,
        monkeypatch,
    ):
-        services = ServiceBundle(
+        services = ServiceBundle.from_session_factory(default_session_factory)
            documents=ServiceBundle().documents.__class__(session_factory=default_session_factory),
            jobs=ServiceBundle().jobs.__class__(session_factory=default_session_factory),
            sources=ServiceBundle().sources.__class__(session_factory=default_session_factory),
            people=ServiceBundle().people.__class__(session_factory=default_session_factory),
        )
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="durability-doc")
            session.add(document)
@@ -155,7 +282,7 @@ class TestWorkflowReliability:
        task = asyncio.create_task(process_queued_job(job=loaded, services=services))
        await asyncio.wait_for(second_started.wait(), timeout=2)
-        attempts = await services.sources.list_execution_attempts(job_id=job.id)
+        attempts = await services.evidence.list_execution_attempts(job_id=job.id)
        assert len(attempts) == 1
        assert attempts[0].raw_transcription == "page 1"
@@ -7,8 +7,8 @@ import pytest
 from pydantic import ValidationError
 from transcription.config import Settings
 from transcription.services.errors import PromptLoadError
 from transcription.services.sources import PromptExecution
 from transcription.services.sources import PromptLoadError
 from transcription.services.sources import build_prompt_execution
 from transcription.services.sources import load_prompt_text
@@ -1,9 +1,10 @@
 """Structural rules for the services package.
-`.github/instructions/services.instructions.md:13` requires that service classes
+The "Structure" section of `.github/instructions/services.instructions.md` requires
-stay independent of one another. Shared behavior belongs in a neutral module
+that service modules stay independent of one another. Shared behavior belongs in a
-(`base.py`, `registry.py`, `source_media.py`, `media_storage.py`), and any
+neutral module that defines no service class (`base.py`, `errors.py`, `registry.py`,
-operation spanning two services belongs in an orchestration module.
+`source_media.py`, `media_storage.py`), and any operation that writes models owned by
 two services belongs in an orchestration module.
 """
 from __future__ import annotations
@@ -13,7 +14,7 @@ from pathlib import Path
 SERVICES_DIR = Path(__file__).resolve().parents[1] / "src" / "transcription" / "services"
-# Modules that intentionally compose several services rather than owning one table.
+# Modules that intentionally compose several services rather than owning one aggregate.
 ORCHESTRATION_MODULES = frozenset({"store", "workflows", "__init__"})
@@ -25,6 +25,7 @@ from transcription.providers.base import TranscriptionResult
 from transcription.providers.evidence import SourceEvidenceReference
 from transcription.providers.openrouter import OpenRouterTranscriptionProvider
 from transcription.services.documents import DocumentService
 from transcription.services.evidence import EvidenceService
 from transcription.services.jobs import JobDeleteBlockedError
 from transcription.services.jobs import JobService
 from transcription.services.sources import SourceService
@@ -203,6 +204,7 @@ async def test_attempts_are_append_only_and_exported_with_integrity(default_sess
    documents = DocumentService(session_factory=default_session_factory)
    jobs = JobService(session_factory=default_session_factory)
    sources = SourceService(session_factory=default_session_factory)
    evidence = EvidenceService(session_factory=default_session_factory)
    document = await documents.create_document(Document(name="Evidence"))
    job = await jobs.create_job(Job(document_id=document.id))
    source = await sources.create_source(
@@ -239,14 +241,14 @@ async def test_attempts_are_append_only_and_exported_with_integrity(default_sess
        finished_at=now,
    )
-    attempts = await sources.list_execution_attempts(source_id=source.id)
+    attempts = await evidence.list_execution_attempts(source_id=source.id)
    assert [attempt.attempt_number for attempt in attempts] == [1, 2]
    assert attempts[0].status == JobSourceStatus.FAILED
    assert attempts[0].error_detail == "first failed"
    assert attempts[1].status == JobSourceStatus.TRANSCRIBED
    assert attempts[1].raw_transcription == "second succeeded"
-    export = await sources.build_evidence_export(source_id=source.id)
+    export = await evidence.build_evidence_export(source_id=source.id)
    assert _json_object(export["source"])["digest_sha256"] == "a" * 64
    assert [_json_object(item)["attempt_number"] for item in _json_array(export["attempts"])] == [1, 2]
    assert "file_path" not in json.dumps(export)
@@ -257,7 +259,7 @@ async def test_attempts_are_append_only_and_exported_with_integrity(default_sess
    latest_job_source = detail.latest_job_source
    assert latest_job_source is not None
    assert latest_job_source.execution_attempts == []
-    latest_attempt = await sources.read_latest_execution_attempt(job_source_id=latest_job_source.id)
+    latest_attempt = await evidence.read_latest_execution_attempt(job_source_id=latest_job_source.id)
    assert latest_attempt is not None
    assert latest_attempt.attempt.attempt_number == 2
@@ -4,6 +4,8 @@ from typing import cast
 import pytest
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 from transcription.services import ServiceBundle
 from transcription.services.sources import SourceService
 from transcription.worker import process_next_queued_job
@@ -11,7 +13,38 @@ from transcription.worker import run_worker_loop
@pytest.mark.asyncio
-async def test_run_worker_loop_survives_process_next_exception(monkeypatch, caplog):
+async def test_run_worker_loop_stops_on_non_retriable_exception(monkeypatch, caplog):
    """A programming error before a job is claimed stops the loop instead of spinning.
    Regression guard for review log [8]. This previously spun at the poll interval
    forever: the fault was classified non-retriable, logged, and then discarded, and
    it never reached the per-job retry machinery so nothing capped it. Measured at 20
    iterations in 1.2s before the fix.
    """
    calls = 0
    stop_event = asyncio.Event()
    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
        nonlocal calls
        _ = (session, session_factory, services)
        calls += 1
        raise RuntimeError("boom")
    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
    with caplog.at_level(logging.CRITICAL):
        await asyncio.wait_for(
            run_worker_loop(stop_event=stop_event, poll_interval_seconds=0),
            timeout=5,
        )
    assert calls == 1
    assert "Worker loop stopped after a non-retriable error" in caplog.text
@pytest.mark.asyncio
 async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog):
    """A retriable fault is still suppressed so transient conditions do not stop work."""
    calls = 0
    stop_event = asyncio.Event()
@@ -20,7 +53,12 @@ async def test_run_worker_loop_survives_process_next_exception(monkeypatch, capl
        _ = (session, session_factory, services)
        calls += 1
        if calls == 1:
-            raise RuntimeError("boom")
+            raise AppError(
                "transient",
                category=ErrorCategory.EXTERNAL_PROVIDER,
                suggestion="retry",
                retriable=True,
            )
        stop_event.set()
        return False
Author	SHA1	Message	Date
zoltan57andCopilot App	edcfba9cb2	Phase 6: enforce the quality gate in CI and export the V4.7 review log Quality Gate / gate (push) Successful in 33s Details Adds .github/workflows/quality-gate.yml, running the gate on push and pull request. CI invokes `pre-commit run --all-files` rather than restating the `ruff check` and `ty check` commands, so the checks keep a single definition in .pre-commit-config.yaml and local and CI cannot drift (plan task 2). The workflow writes a .env file rather than exporting an environment variable. The two are not equivalent here: Settings reads the .env file, while the external-test skip guard reads os.getenv, so an exported variable un-skips the external tests and sends them to the network. Measured in CI: no .env gave 115 failures and 18 errors, an exported dummy key gave 3 failures, and a written .env file reproduced the local baseline exactly. Negative-tested on a scratch branch: a deliberate lint error failed the run at `ruff check` with exactly the planted errors, confirming the gate blocks rather than merely reporting (plan task 4). The subsequent clean run passed ruff and ty and reported 295 passed, 4 skipped, matching local and confirming the four credential-gated tests skip cleanly (plan task 3). That first green run caught a real platform-dependent defect. PromptStore rejected non-direct-child names via `Path(name).name != name`, which is platform-dependent: on POSIX a backslash is an ordinary filename character, so "nested\prompt.md" passed the guard and failed later as NOT_FOUND rather than VALIDATION. Windows cannot reproduce it. No traversal was possible, since the path.parent != root check still held, so the impact was a wrong error category and a red gate. Both separators are now rejected explicitly, matching the ^[^/\\]+$ pattern config.PromptFilename already used. Also exports docs/ver4.7/review_log_v4_7.md, the working record kept across all six phases: 50 entries, 1 still open. The open entry is a pre-existing /ui redirect defect found during the Phase 3 UI walk and deliberately left unfixed as outside the V4.7 scope boundary. Co-authored-by: Copilot App <[email protected]>	2026-08-18 23:05:52 -05:00
zoltan57andCopilot App	fca959fa5d	Phase 5: contain worker faults instead of discarding the classification Review log [8]. classify_unexpected_error already returned retriable=False and the verdict was logged and then thrown away. Measured across src/: retriable was assigned in 9 places and read in none. The plan asks for a test that a programming error "does not silently retry". Probing with an injected AttributeError showed that is not what happens, and the two real failure modes need different fixes. Mode A, raised after the claim commits (inside advance_job): raised exactly once, job left at PROCESSING, retry_count 0, never re-claimed, because claim_next_queued_job filters status == QUEUED. A permanently stranded job with one swallowed log line, not a retry. advance_job's PROCESSING branch, commented "Recover mid-flight jobs", is unreachable from the worker for the same reason. Mode B, raised before or during the claim: 20 raises in 1.2s, an unbounded hot spin at the poll interval. It never reaches the per-job retry machinery, so WORKER_MAX_RETRIES does not cap it and the plan's 60s worst case understates this path. services/workflows.py _advance_job_with_containment wraps advance_job. Any escaping exception is classified and the job driven to terminal FAILED, which is visible in the UI and resubmittable. The caller session is rolled back first and the terminal write runs in its own transaction, so it stays atomic even when the failure left that session dirty (plan task 3). The loop continues, so one poison job cannot halt transcription for every other job. worker.py handle_worker_exceptions re-raises non-retriable faults rather than suppressing them; retriable ones are still suppressed so transient conditions do not stop work. run_worker_loop catches that, logs CRITICAL and returns cleanly. Returning rather than propagating matters: the exception would otherwise surface only at app shutdown, through the wait_for in worker_consumer_lifespan. tests test_run_worker_loop_survives_process_next_exception asserted the loop SURVIVES a RuntimeError and continues, which is the Mode B defect written down as an expectation. Replaced by test_run_worker_loop_stops_on_non_retriable_exception, with a new test_run_worker_loop_survives_retriable_exception so suppression of genuinely transient faults stays covered, and test_error_after_claim_fails_the_job_instead_of_stranding_it for Mode A. All three were verified to fail on pre-fix code. The Mode B guard fails by timing out, which is the infinite spin made visible. Verified: 295 passed, 4 skipped, 0 ruff, 0 ty. Co-authored-by: Copilot App <[email protected]>	2026-08-18 16:12:52 -05:00
zoltan57andCopilot App	110f40a28b	Phase 4: measure only the provider call in duration_ms Review log [55]. Three historical local_timeout rows recorded 0.4-2.0s more than the configured budget because the measurement window opened before the provider call. The plan named two causes, and both were already gone. Diffed against f86c0ff~1: at V4.6 the window held resolve_provider_input (async; normalization + artifact write + DB work) and a session.commit(). Phase 1 deleted both. What remains between the clock and the wait_for is build_provider_input, now pure field copying because normalization moved to ingest and file_hash is already stored: 6.2 us per call, zero awaits, so it cannot yield to the event loop. A third cause was still there and is not in the plan. The regression test below measured 890ms where ~200ms was expected. services.sources.provider is a lazy property that appears as an argument expression to _call_transcriber, so it is evaluated after the clock starts but before wait_for begins timing. Constructing OpenRouterTranscriptionProvider costs 475ms on first access and 0.001ms after, so the first attempt of every worker process booked half a second of HTTP client construction as provider latency. That plausibly accounts for the low end of the historical overshoot. workflows.py - Re-capture monotonic_started_at immediately before the wait_for, reusing the same variable. The pre-loop assignment stays as the fallback: binding a new name inside the try would leave the general-exception handler referencing an unbound variable when build_provider_input raises. All three duration write sites (success, TimeoutError, general failure) then measure the correct window with no further change. - Hoist the provider property above the per-source loop. It is loop-invariant, so this also removes the repeated lookup from the two evidence-capture sites. tests/services/test_workflows_reliability.py test_timeout_duration_excludes_pre_call_setup simulates 400ms of blocking setup against a 200ms budget and asserts the recorded duration sits near the budget and well clear of budget+setup. Confirmed to fail on the pre-fix code (assert 625 < 540) and pass after, so it guards behaviour rather than restating it. This is the plan's verification criterion as a test. ui/pages/sources_page.py _format_duration renders >=1s as "27.6 s" and below that as "612 ms", replacing the raw "27612 ms". No test asserted the old format. Plan task 3 (record preprocessing as its own value) declined and logged as a deviation: after Phase 1 there is no preprocessing left to record, and a preprocessing_ms column to measure 6 us of attribute copying is complexity without a reader. Verified: 293 passed, 4 skipped, 0 ruff, 0 ty. Co-authored-by: Copilot App <[email protected]>	2026-08-18 16:02:58 -05:00
zoltan57andCopilot App	7dd0d2c9bf	Phase 3: extract EvidenceService and rewrite the service ownership rule Decompose SourceService along the aggregate boundary and then correct the instruction file that caused it to grow, in that order. The refactor is the empirical test of the rule. services/evidence.py (new) EvidenceService owns ExecutionAttempt: read_latest_execution_attempt, list_execution_attempts, promote_machine_attempt, build_evidence_export, plus the LatestExecutionAttempt projection. Moved verbatim from sources.py. services/errors.py (new) The five-class error hierarchy (PromptLoadError, TranscriptionError, TranscriptionNotFoundError, SourceDeleteBlockedError, CandidatePromotionError) moved out of sources.py. evidence.py needs TranscriptionNotFoundError, and test_service_boundaries.py correctly rejected the sibling import. errors.py defines no Service class, so it is a legal shared home. This was the boundary test doing its job, not an obstacle to route around. sources.py 1,389 -> 885 lines (1,063 after Phase 2). services/__init__.py ServiceBundle and from_session_factory register evidence. Note that field-by-field ServiceBundle construction silently binds services to the process-global session factory via default_factory; from_session_factory is the only safe constructor. Two test bundles were fixed for this. .github/instructions/services.instructions.md Rewritten to describe the boundaries the decomposition actually produced, per plan Phase 3 task 7 and review log [59]. - "1 service class per data model" -> one service class per aggregate. The table-shaped rule is the measured cause of sources.py reaching 1,389 lines; DocumentType has no lifecycle without Document. - New Model Ownership section. Junctions are owned by their lifecycle owner, the service that creates and deletes the rows: document_person to PeopleService (sole writer, measured), job_source to SourceService. Two carve-outs are stated rather than left as silent violations: cascade deletion when a service deletes its own aggregate root, and status transitions that create and delete nothing (cancel_job, resubmit_failed_sources), which are Job lifecycle events on the work queue. EvidenceService.promote_machine_attempt's two-field write to Source is named and scoped. - Mandatory CRUD softened to intent. It was already false: five modules define no service class, EvidenceService has no create/delete because ExecutionAttempt is append-only, RegistryService uses <op>_entry. - Separated reading across models via eager loads from the owning root, which is allowed, from importing another service, which is not. The old line 13 and lines 75-77 read as contradictory. - Typo: picutre. No code was moved to satisfy the rule. tests/test_service_boundaries.py Docstring no longer cites the instruction file by line number; that anchor would desynchronise silently. errors.py added to the neutral-module list. Verified: 292 passed, 4 skipped, 0 ruff, 0 ty. All 25 /ui/ routes walked against the live app; 24x 200. /ui/documents/{id}/sources 404s via a 307 that drops the /ui prefix, confirmed pre-existing (last touched in `6a3ee26`) and left alone as out of scope. Co-authored-by: Copilot App <[email protected]>	2026-08-18 15:54:27 -05:00