generated from john/python-template
Compare commits
4
Commits
11097b9cfe
...
edcfba9cb2
| Author | SHA1 | Date | |
|---|---|---|---|
  zoltan57andCopilot App
|
edcfba9cb2 | ||
|
zoltan57andCopilot App
|
fca959fa5d | ||
|
zoltan57andCopilot App
|
110f40a28b | ||
|
zoltan57andCopilot App
|
7dd0d2c9bf |
@@ -7,29 +7,89 @@ applyTo: 'src/transcription/services/*.py'
|
||||
|
||||
## Structure
|
||||
|
||||
- Project core data models defined in [models](../../src/transcription/db/models.py)
|
||||
- 1 service class per data model
|
||||
- Only services directly interact with the database, and only through async methods
|
||||
- Services are completely independent of one another. Any operation that needs to use more than a single service, which is most of them, needs to have a separate orchestration function.
|
||||
- Project core data models are defined in [models](../../src/transcription/db/models.py)
|
||||
- One service class per **aggregate**, not per table. An aggregate is a root model plus
|
||||
the models that have no independent lifecycle of their own. `DocumentType` has no
|
||||
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`
|
||||
- Use a context manager for large `try/except` blocks like `handle_transcription_errors` in [sources](../../src/transcription/services/sources.py)
|
||||
- Errors used by a single service are defined at the top of that module and inherit from `AppError`.
|
||||
- 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 in a session object to each method
|
||||
- [ ] Services use `self._session_scope` in their methods to pass the session thru.
|
||||
- Multiple operations on the same object(s) require sharing a session between all the methods used.
|
||||
- [ ] Session kwarg for `AsyncSession` to pass a session object into each method
|
||||
- [ ] Services use `self._session_scope` in their methods to pass the session through
|
||||
- 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`
|
||||
- All services must define these 4 methods first, and in that order
|
||||
- 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
|
||||
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,
|
||||
transport_evidence=services.sources.provider.current_transport_evidence,
|
||||
request_manifest=provider.current_request_manifest,
|
||||
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.sources import SourceDeleteBlockedError
|
||||
from transcription.services.errors 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.jobs import JobService
|
||||
from transcription.services.sources import CandidatePromotionError
|
||||
from transcription.services.sources import SourceService
|
||||
from transcription.services.errors import CandidatePromotionError
|
||||
from transcription.services.workflows import create_source_retranscription_job
|
||||
|
||||
|
||||
def _services(default_session_factory, settings: Settings) -> ServiceBundle:
|
||||
return ServiceBundle(
|
||||
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),
|
||||
)
|
||||
return ServiceBundle.from_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(
|
||||
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),
|
||||
)
|
||||
services = ServiceBundle.from_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
|
||||
stay independent of one another. Shared behavior belongs in a neutral module
|
||||
(`base.py`, `registry.py`, `source_media.py`, `media_storage.py`), and any
|
||||
operation spanning two services belongs in an orchestration module.
|
||||
The "Structure" section of `.github/instructions/services.instructions.md` requires
|
||||
that service modules stay independent of one another. Shared behavior belongs in a
|
||||
neutral module that defines no service class (`base.py`, `errors.py`, `registry.py`,
|
||||
`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
|
||||
|
||||
|
||||
+40
-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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user