diff --git a/docs/ui/README.md b/docs/ui/README.md index d1a2e14..8f04634 100644 --- a/docs/ui/README.md +++ b/docs/ui/README.md @@ -57,4 +57,5 @@ Each page contract contains: ## Current Baseline -These contracts describe the V4 baseline with completed V4.1 UI behavior. Planned V4.2 evidence/provenance changes and draft V4.3 Settings/page-reordering changes are not described as current behavior. +These contracts describe the V4 baseline with completed V4.1 behavior and V4.2 evidence/provenance behavior. +Draft V4.3 Settings/page-reordering changes are not described as current behavior. diff --git a/docs/ui/pages/jobs.md b/docs/ui/pages/jobs.md index c54ef9f..36b4c53 100644 --- a/docs/ui/pages/jobs.md +++ b/docs/ui/pages/jobs.md @@ -58,9 +58,10 @@ Jobs manages transcription processing runs. A Job belongs to one Document, links ## Resubmit Behavior - The page shows current status and failed Source count. -- The page explicitly states: `Resubmit queues only failed linked sources. New results overwrite prior page-level results.` +- The page explains that resubmission queues failed linked Sources while preserving immutable prior attempt evidence. - The service blocks submission while processing is active or when no failed Sources exist. -- Current behavior updates the existing page-level result when new output arrives. +- `JobSource` remains the latest compatibility projection, while every provider call appends an `ExecutionAttempt`. +- The latest successful `Source.raw_transcription` projection remains available while a retry is pending or fails. - Success reports the number of resubmitted Sources and returns to Job Detail. ## Delete Behavior @@ -88,7 +89,3 @@ Jobs manages transcription processing runs. A Job belongs to one Document, links - `tests/ui/test_jobs_page.py` - `tests/services/test_job_service.py` - `tests/services/test_store.py` - -## Planned Change - -V4.2 replaces update-in-place retry evidence with append-only processing attempts and adds exact transport evidence. Until implemented, the current overwrite behavior must be labeled accurately rather than described as archival history. See the [V4.2 scope](../../ver4.2/scope_boundary_v4_2.md). diff --git a/docs/ui/pages/sources.md b/docs/ui/pages/sources.md index a5de189..35c77de 100644 --- a/docs/ui/pages/sources.md +++ b/docs/ui/pages/sources.md @@ -42,9 +42,13 @@ The list accepts optional `document_id` and `job_id` query parameters. Document ## Provider Evidence - Provider Evidence is associated with the latest JobSource execution. -- AI Metadata and the current `raw_api_response` value are displayed as expandable formatted JSON. +- New attempts display separate expandable Request Manifest, Transport Response, OpenRouter SDK Response Snapshot, + Normalized Metadata, Software Context, and Derived Artifacts sections. +- Historical `raw_api_response` values are labeled as OpenRouter SDK response snapshots. - Missing evidence has an explicit empty state. -- Under the current V4 implementation, `raw_api_response` is an OpenRouter SDK response snapshot, not an exact HTTP or native upstream-provider response. +- Historical executions explicitly state that exact transport evidence was not captured. +- **Export Evidence** downloads a versioned package containing source identity, attempts, artifacts, relationships, + schema versions, and integrity digests without source binaries, credentials, or machine-local source paths. ## Revision Behavior @@ -82,5 +86,4 @@ The list accepts optional `document_id` and `job_id` query parameters. Document ## Planned Changes -- V4.2 will rename and separate evidence layers, add exact OpenRouter transport capture, and preserve append-only attempts. See the [V4.2 scope](../../ver4.2/scope_boundary_v4_2.md). - Source page reordering is deferred to the [draft V4.3 scope](../../ver4.3/scope_boundary_v4_3.md). diff --git a/docs/ver4.2/implementation_plan_v4_2.md b/docs/ver4.2/implementation_plan_v4_2.md index 5300e69..9076f5d 100644 --- a/docs/ver4.2/implementation_plan_v4_2.md +++ b/docs/ver4.2/implementation_plan_v4_2.md @@ -210,6 +210,23 @@ Enforce exactly one content location: inline payload or external reference. An e These decisions must be settled before their corresponding implementation phase; they do not weaken the invariant or expand V4.2 into live OCR integration. +## Resolved Implementation Decisions + +1. `JobSource` remains queue linkage and a compatibility projection; immutable retries use a one-to-many + `ExecutionAttempt` model with a unique `(job_id, source_id, attempt_number)` constraint. +2. Exact OpenRouter response bytes remain database values for V4.2. Generic artifacts use inline canonical JSON up + to 1 MiB by default and constrained, atomically written external files above that threshold. +3. Request manifests use `transcription-canonical-json-v1`: UTF-8 JSON with sorted keys, compact separators, + preserved Unicode, and non-finite numbers rejected. +4. Safe response headers are explicitly allowlisted in the evidence contract; all others are discarded before + persistence. +5. Software identity records the package version, optional `TRANSCRIPTION_COMMIT`, adapter contract version, + OpenRouter SDK version, and Python version. +6. The artifact root defaults to `data/artifacts` and stores source-scoped relative references. +7. Evidence exports include source identity and digest by reference, not original source binaries. +8. The benchmark manifest is private and digest-referenced. Corpus size remains archive-dependent, but every run + uses preserved execution-attempt identity and the fixed literal scoring contract. + ## Done When - Every V4.2 acceptance criterion is satisfied by focused tests or an explicit demonstration. diff --git a/docs/ver4/architecture_v4.md b/docs/ver4/architecture_v4.md index 91c1b66..3888258 100644 --- a/docs/ver4/architecture_v4.md +++ b/docs/ver4/architecture_v4.md @@ -26,7 +26,10 @@ This document describes the production architecture of the document transcriptio - Isolate page failures so multi-page jobs can complete with partial success. - Operate across supported platforms through Python-based application and maintenance tooling. -V4.2 extends this baseline with exact OpenRouter transport evidence and provider-neutral derived-artifact provenance. See the [V4.2 Scope Boundary](../ver4.2/scope_boundary_v4_2.md). +V4.2 extends this baseline with immutable execution attempts, exact OpenRouter transport evidence, safe +versioned exports, and provider-neutral derived-artifact provenance. `JobSource` remains the mutable queue and +compatibility projection; `ExecutionAttempt` is the authoritative append-only processing history. See the +[V4.2 Scope Boundary](../ver4.2/scope_boundary_v4_2.md). ## Technical Stack @@ -135,8 +138,11 @@ Responsibilities: 1. User uploads one or more images for a `Document`. 2. System stores files, hashes them, creates ordered `Source` rows, and creates a `Job`. 3. Worker claims the job, marks it `processing`, and executes page calls concurrently. -4. Each page writes a `JobSource` result with machine output, normalized metadata, and an SDK-serialized OpenRouter response snapshot. -5. Aggregate status becomes `completed`, `partial_success`, or `failed`. +4. Each provider call appends an `ExecutionAttempt` with its request manifest, transport evidence, SDK snapshot, + normalized metadata, timing, and outcome. +5. The linked `JobSource` is updated as a compatibility projection, and a successful attempt updates the + `Source.raw_transcription` latest-success projection. +6. Aggregate status becomes `completed`, `partial_success`, or `failed`. ### 2. Document-Person Relationship Management @@ -160,6 +166,9 @@ Responsibilities: - Human corrections occur only in `Source.revised_text`. - Prompt and parameter provenance is frozen on `Job` at submission time. - The SDK-serialized OpenRouter response snapshot is stored on `JobSource` for each successful page execution. +- Every V4.2 provider call appends a distinct `ExecutionAttempt`; retries never rewrite earlier attempts. +- Exact response bytes identify the OpenRouter HTTP boundary and are not labeled as native upstream-provider JSON. +- Generic `ProcessingArtifact` records use versioned schemas, digests, and one inline or external content location. - `DocumentPerson` links are unique for `(document_id, person_id, role_id)`. - Relationship mutations are deterministic and set-based. - `DocumentType.code` is stable; `DocumentType.label` may evolve. diff --git a/docs/ver4/requirements_v4.md b/docs/ver4/requirements_v4.md index 3292df8..03a3890 100644 --- a/docs/ver4/requirements_v4.md +++ b/docs/ver4/requirements_v4.md @@ -9,7 +9,7 @@ This document defines the baseline requirements for the document transcription s | REQ-0 | System | Provide end-to-end multi-page document transcription with persistent, inspectable async job states. | demonstration | | REQ-1 | Functional | Allow users to upload one or more images as ordered `Source` pages under a `Document`. | test | | REQ-2 | Functional | Process page transcription asynchronously using an `asyncio` worker pool bounded by rate limits. | test | -| REQ-3 | Functional | Persist submission-time prompt configuration and full page-level provider response evidence for every job execution. | test | +| REQ-3 | Functional | Persist submission-time request provenance and accurately labeled page-level SDK evidence; V4.2 adds exact OpenRouter-boundary transport evidence for new attempts. | test | | REQ-4 | Functional | Support job states `queued`, `processing`, `completed`, `partial_success`, and `failed`, plus page states `pending`, `transcribed`, and `failed`. | inspection | | REQ-5 | Functional | Allow users to manage historical `Person` records and link multiple people per role to a `Document`. | test | | REQ-6 | Functional | Support an extensible role taxonomy for document-person relationships. | inspection | diff --git a/docs/ver4/schema_v4.md b/docs/ver4/schema_v4.md index 359f9e0..ddd9563 100644 --- a/docs/ver4/schema_v4.md +++ b/docs/ver4/schema_v4.md @@ -109,6 +109,46 @@ TEXT error_detail TIMESTAMPTZ executed_at } +EXECUTION_ATTEMPT { +UUID id PK +UUID job_source_id FK +UUID job_id FK +UUID source_id FK +INTEGER attempt_number +VARCHAR status +JSONB request_manifest +TEXT request_manifest_sha256 +INTEGER transport_status_code +BINARY transport_body +JSONB transport_safe_headers +JSONB sdk_response_snapshot +JSONB normalized_metadata +JSONB software_context +TEXT raw_transcription +TEXT failure_phase +TIMESTAMPTZ started_at +TIMESTAMPTZ finished_at +INTEGER duration_ms +} + +PROCESSING_ARTIFACT { +UUID id PK +UUID source_id FK +UUID execution_attempt_id FK +TEXT artifact_type +TEXT media_type +TEXT schema_name +TEXT schema_version +TEXT producer +TEXT producer_version +JSONB inline_payload +TEXT external_reference +TEXT payload_sha256 +BIGINT byte_size +JSONB coordinate_metadata +TIMESTAMPTZ created_at +} + DOCUMENT_TYPE ||--o{ DOCUMENT : classifies DOCUMENT ||--o{ DOCUMENT_PERSON : has_people PERSON ||--o{ DOCUMENT_PERSON : appears_in @@ -117,6 +157,9 @@ DOCUMENT ||--o{ JOB : has_jobs DOCUMENT ||--o{ SOURCE : contains_pages JOB ||--o{ JOB_SOURCE : executes SOURCE ||--o{ JOB_SOURCE : processed_in +JOB_SOURCE ||--o{ EXECUTION_ATTEMPT : projects +SOURCE ||--o{ PROCESSING_ARTIFACT : derives +EXECUTION_ATTEMPT ||--o{ PROCESSING_ARTIFACT : produces ``` ## Domain Invariants and Provenance Rules @@ -125,9 +168,23 @@ SOURCE ||--o{ JOB_SOURCE : processed_in - Every single page execution by an AI model produces a dedicated `JOB_SOURCE` record. - Every `JOB` stores the frozen prompt identifier, prompt text, and hyperparameters used at submission time. -- Every `JOB_SOURCE` stores the complete provider response envelope and page-level operational metadata. +- `JOB_SOURCE.raw_api_response` is a compatibility projection containing an SDK-serialized OpenRouter response + snapshot. It is neither the exact HTTP body nor the native upstream-provider response. +- Every new provider call creates an immutable `EXECUTION_ATTEMPT` containing the frozen request manifest, + exact OpenRouter-boundary response bytes when received, safe transport metadata, SDK snapshot, normalized + metadata, timing, and outcome. +- `EXECUTION_ATTEMPT(job_id, source_id, attempt_number)` is unique; retries increment the persisted attempt number. +- Historical `JOB_SOURCE` rows without an `EXECUTION_ATTEMPT` remain SDK snapshots and are explicitly labeled as + lacking transport evidence. - `SOURCE.raw_transcription` caches the latest successful machine output for that page. +### Generic Processing Artifacts + +- `PROCESSING_ARTIFACT` stores provider-neutral versioned derived outputs. +- Exactly one of `inline_payload` and `external_reference` is populated. +- Externally stored artifacts use application-managed relative references and are verified by SHA-256 and byte size. +- Coordinate metadata declares units, origin, dimensions, and transformations when geometry is present. + ### Image Storage and Integrity - Binary images are stored on disk; `SOURCE.file_path` stores the persisted path. diff --git a/prompts/transcribe_document.md b/prompts/transcribe_document.md index c136e57..c8b7740 100644 --- a/prompts/transcribe_document.md +++ b/prompts/transcribe_document.md @@ -46,6 +46,16 @@ Do not summarize. Do not paraphrase. Do not modernize style. - Signal location before the note text. - Example form: `[written in left margin: ...]` +### Printed and handwritten text +- Preserve printed and handwritten text together in their original reading context. +- On mixed documents such as completed forms, leave printed labels and instructions unmarked. +- Wrap handwritten entries in `[handwritten: ...]`. +- Mark handwritten signatures as `[handwritten signature: ...]`. +- If the main body is entirely handwritten, add `[document body handwritten]` once at the beginning rather than marking every line. +- Mark later notes or uncertain additions as `[handwritten annotation: ...]`. +- When authorship is unclear, use `[handwritten annotation, author uncertain: ...]`. +- Do not infer authorship, writing date, or whether different handwriting belongs to different people unless explicitly evident. + ### Line-break hyphenation - Rejoin words split across line breaks when they are clearly one word. - Remove only line-break hyphens used for wrapping. @@ -70,3 +80,4 @@ Before finalizing, ensure: 2. Uncertain/illegible areas are explicitly marked. 3. Crossed-out and inserted text are preserved with required tags. 4. Structure/ordering is preserved as faithfully as possible. +5. Handwriting is identified using the mixed-text conventions without separating it from its printed context. diff --git a/src/transcription/benchmarking.py b/src/transcription/benchmarking.py new file mode 100644 index 0000000..9f7c22b --- /dev/null +++ b/src/transcription/benchmarking.py @@ -0,0 +1,102 @@ +"""Private-corpus benchmark contracts and deterministic text scoring.""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field + + +class BenchmarkModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class BenchmarkItem(BenchmarkModel): + """One private benchmark item referenced by archival identity.""" + + source_id: UUID + source_digest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + categories: frozenset[str] = Field(min_length=1) + reference_transcription: str = Field(min_length=1) + + +class BenchmarkManifest(BenchmarkModel): + """Versioned private benchmark definition without copied source media.""" + + schema_name: str = "transcription.private-benchmark" + schema_version: str = "1" + name: str = Field(min_length=1) + items: tuple[BenchmarkItem, ...] = Field(min_length=1) + + +class EditorialAssessment(BenchmarkModel): + """Manually reviewed errors not represented adequately by CER or WER.""" + + omissions: int = Field(default=0, ge=0) + inventions: int = Field(default=0, ge=0) + silent_normalizations: int = Field(default=0, ge=0) + uncertainty_errors: int = Field(default=0, ge=0) + layout_errors: int = Field(default=0, ge=0) + + +class BenchmarkScore(BenchmarkModel): + """Measured score for one preserved execution attempt.""" + + execution_attempt_id: UUID + character_error_rate: float = Field(ge=0) + word_error_rate: float = Field(ge=0) + character_edits: int = Field(ge=0) + word_edits: int = Field(ge=0) + reference_characters: int = Field(ge=0) + reference_words: int = Field(ge=0) + assessment: EditorialAssessment + latency_ms: int = Field(ge=0) + cost_usd: float | None = Field(default=None, ge=0) + + +def score_transcription( + *, + execution_attempt_id: UUID, + reference: str, + candidate: str, + assessment: EditorialAssessment, + latency_ms: int, + cost_usd: float | None = None, +) -> BenchmarkScore: + """Score literal text without case-folding or silent normalization.""" + reference_words = reference.split() + candidate_words = candidate.split() + character_edits = _levenshtein(list(reference), list(candidate)) + word_edits = _levenshtein(reference_words, candidate_words) + return BenchmarkScore( + execution_attempt_id=execution_attempt_id, + character_error_rate=character_edits / max(1, len(reference)), + word_error_rate=word_edits / max(1, len(reference_words)), + character_edits=character_edits, + word_edits=word_edits, + reference_characters=len(reference), + reference_words=len(reference_words), + assessment=assessment, + latency_ms=latency_ms, + cost_usd=cost_usd, + ) + + +def _levenshtein(reference: list[str], candidate: list[str]) -> int: + if len(reference) < len(candidate): + reference, candidate = candidate, reference + previous = list(range(len(candidate) + 1)) + for reference_index, reference_value in enumerate(reference, start=1): + current = [reference_index] + for candidate_index, candidate_value in enumerate(candidate, start=1): + current.append( + min( + current[-1] + 1, + previous[candidate_index] + 1, + previous[candidate_index - 1] + (reference_value != candidate_value), + ) + ) + previous = current + return previous[-1] diff --git a/src/transcription/config.py b/src/transcription/config.py index e546563..1c72a12 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -97,6 +97,8 @@ class Settings(BaseSettings): # --- filesystem paths --- upload_dir: Path = Path("./uploads") prompt_dir: Path = Path("./prompts") + artifact_dir: Path = Path("./data/artifacts") + artifact_inline_threshold_bytes: int = Field(default=1_048_576, ge=1) # --- worker reliability --- worker_max_retries: int = Field(default=0, ge=0) diff --git a/src/transcription/db/models.py b/src/transcription/db/models.py index c3068d8..c4f3de3 100644 --- a/src/transcription/db/models.py +++ b/src/transcription/db/models.py @@ -11,8 +11,10 @@ from uuid import uuid4 from pydantic import JsonValue from sqlalchemy import JSON from sqlalchemy import BigInteger +from sqlalchemy import CheckConstraint from sqlalchemy import Column from sqlalchemy import Enum as SAEnum +from sqlalchemy import LargeBinary from sqlalchemy import UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm.exc import DetachedInstanceError @@ -25,12 +27,12 @@ from sqlmodel import SQLModel class JSONBCompat(TypeDecorator): """JSONB for PostgreSQL and JSON for SQLite/testing backends.""" - impl = JSON + impl = JSON(none_as_null=True) def load_dialect_impl(self, dialect): if dialect.name == "postgresql": - return dialect.type_descriptor(JSONB()) - return dialect.type_descriptor(JSON()) + return dialect.type_descriptor(JSONB(none_as_null=True)) + return dialect.type_descriptor(JSON(none_as_null=True)) class JobStatus(StrEnum): @@ -270,6 +272,10 @@ class Source(SQLModel, table=True): back_populates="source", sa_relationship_kwargs={"lazy": "selectin"}, ) + processing_artifacts: list["ProcessingArtifact"] = Relationship( + back_populates="source", + sa_relationship_kwargs={"lazy": "noload"}, + ) @property def latest_job_source(self) -> Optional["JobSource"]: @@ -323,3 +329,90 @@ class JobSource(SQLModel, table=True): job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) + execution_attempts: list["ExecutionAttempt"] = Relationship( + back_populates="job_source", + sa_relationship_kwargs={"lazy": "noload", "order_by": "ExecutionAttempt.attempt_number"}, + ) + + +class ExecutionAttempt(SQLModel, table=True): + """Immutable evidence for one provider call attempt.""" + + __tablename__ = "execution_attempt" + __table_args__ = (UniqueConstraint("job_id", "source_id", "attempt_number", name="uq_execution_attempt_number"),) + + id: UUID = Field(default_factory=uuid4, primary_key=True) + job_source_id: UUID = Field(foreign_key="job_source.id", index=True) + job_id: UUID = Field(foreign_key="job.id", index=True) + source_id: UUID = Field(foreign_key="source.id", index=True) + attempt_number: int = Field(ge=1) + status: JobSourceStatus + provider: str + model: str | None = None + request_manifest: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) + request_manifest_sha256: str | None = None + request_manifest_schema_version: str | None = None + response_received: bool = False + transport_status_code: int | None = None + transport_body: bytes | None = Field(default=None, sa_column=Column(LargeBinary(), nullable=True)) + transport_content_type: str | None = None + transport_content_encoding: str | None = None + transport_safe_headers: dict[str, JsonValue] | None = Field( + default=None, sa_column=Column(JSONBCompat(), nullable=True) + ) + router_request_id: str | None = None + router_generation_id: str | None = None + sdk_response_snapshot: dict[str, JsonValue] | None = Field( + default=None, sa_column=Column(JSONBCompat(), nullable=True) + ) + normalized_metadata: dict[str, JsonValue] | None = Field( + default=None, sa_column=Column(JSONBCompat(), nullable=True) + ) + software_context: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) + raw_transcription: str | None = None + error_category: str | None = None + error_detail: str | None = None + failure_phase: str | None = None + started_at: datetime + finished_at: datetime + duration_ms: int = Field(ge=0) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + + job_source: Optional["JobSource"] = Relationship(back_populates="execution_attempts") + artifacts: list["ProcessingArtifact"] = Relationship( + back_populates="execution_attempt", sa_relationship_kwargs={"lazy": "noload"} + ) + + +class ProcessingArtifact(SQLModel, table=True): + """Provider-neutral, versioned output derived from a Source.""" + + __tablename__ = "processing_artifact" + __table_args__ = ( + CheckConstraint( + "(inline_payload IS NOT NULL AND external_reference IS NULL) OR " + "(inline_payload IS NULL AND external_reference IS NOT NULL)", + name="ck_processing_artifact_one_content_location", + ), + ) + + id: UUID = Field(default_factory=uuid4, primary_key=True) + source_id: UUID = Field(foreign_key="source.id", index=True) + execution_attempt_id: UUID | None = Field(default=None, foreign_key="execution_attempt.id", index=True) + artifact_type: str + media_type: str + schema_name: str + schema_version: str + producer: str + producer_version: str + inline_payload: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True)) + external_reference: str | None = None + payload_sha256: str = Field(index=True) + byte_size: int = Field(sa_column=Column(BigInteger(), nullable=False)) + coordinate_metadata: dict[str, JsonValue] | None = Field( + default=None, sa_column=Column(JSONBCompat(), nullable=True) + ) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + + execution_attempt: Optional["ExecutionAttempt"] = Relationship(back_populates="artifacts") + source: Optional["Source"] = Relationship(back_populates="processing_artifacts") diff --git a/src/transcription/db/operations.py b/src/transcription/db/operations.py index 1c889bd..b4bee09 100644 --- a/src/transcription/db/operations.py +++ b/src/transcription/db/operations.py @@ -45,6 +45,7 @@ async def create_all(*, engine: AsyncEngine | None = None) -> None: async with active_engine.begin() as connection: await connection.run_sync(SQLModel.metadata.create_all) await _upgrade_person_family_search_id(connection) + await _upgrade_v42_evidence_tables(connection) await seed_registry_defaults(engine=active_engine) logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url) @@ -54,6 +55,17 @@ async def upgrade_schema(*, engine: AsyncEngine | None = None) -> None: active_engine = engine or resolve_engine() async with active_engine.begin() as connection: await _upgrade_person_family_search_id(connection) + await _upgrade_v42_evidence_tables(connection) + + +async def _upgrade_v42_evidence_tables(connection: AsyncConnection) -> None: + """Create the additive V4.2 evidence tables without rewriting historical rows.""" + + def create_tables(sync_connection) -> None: + SQLModel.metadata.tables["execution_attempt"].create(sync_connection, checkfirst=True) + SQLModel.metadata.tables["processing_artifact"].create(sync_connection, checkfirst=True) + + await connection.run_sync(create_tables) async def _upgrade_person_family_search_id(connection: AsyncConnection) -> None: @@ -66,9 +78,7 @@ async def _upgrade_person_family_search_id(connection: AsyncConnection) -> None: columns = {column["name"] for column in database.get_columns("person")} indexes = database.get_indexes("person") constraints = database.get_unique_constraints("person") - has_unique_id = any( - entry.get("column_names") == ["family_search_id"] for entry in [*indexes, *constraints] - ) + has_unique_id = any(entry.get("column_names") == ["family_search_id"] for entry in [*indexes, *constraints]) return "family_search_id" in columns, has_unique_id has_column, has_unique_id = await connection.run_sync(inspect_person) diff --git a/src/transcription/providers/__init__.py b/src/transcription/providers/__init__.py index 2463e48..ff917c7 100644 --- a/src/transcription/providers/__init__.py +++ b/src/transcription/providers/__init__.py @@ -9,6 +9,9 @@ from transcription.providers.base import ProviderResponseError from transcription.providers.base import TranscriptionMetadata from transcription.providers.base import TranscriptionProvider from transcription.providers.base import TranscriptionResult +from transcription.providers.evidence import RequestManifest +from transcription.providers.evidence import SourceEvidenceReference +from transcription.providers.evidence import TransportEvidence from transcription.providers.openrouter import OpenRouterTranscriptionProvider @@ -26,8 +29,11 @@ __all__ = [ "ProviderAuthError", "ProviderError", "ProviderResponseError", + "RequestManifest", + "SourceEvidenceReference", "TranscriptionMetadata", "TranscriptionProvider", "TranscriptionResult", + "TransportEvidence", "get_transcription_provider", ] diff --git a/src/transcription/providers/base.py b/src/transcription/providers/base.py index 2488738..3ad2983 100644 --- a/src/transcription/providers/base.py +++ b/src/transcription/providers/base.py @@ -7,10 +7,27 @@ from pydantic import ConfigDict from pydantic import Field from pydantic import JsonValue +from transcription.providers.evidence import RequestManifest +from transcription.providers.evidence import SourceEvidenceReference +from transcription.providers.evidence import TransportEvidence + class ProviderError(RuntimeError): """Base error for provider failures.""" + def __init__( + self, + message: str, + *, + request_manifest: RequestManifest | None = None, + transport_evidence: TransportEvidence | None = None, + failure_phase: str = "provider_request", + ) -> None: + super().__init__(message) + self.request_manifest = request_manifest + self.transport_evidence = transport_evidence + self.failure_phase = failure_phase + class ProviderAuthError(ProviderError): """Raised when provider authentication fails.""" @@ -59,6 +76,8 @@ class TranscriptionResult(BaseModel): top_p: float | None = Field(default=None, ge=0.0, le=1.0) metadata: TranscriptionMetadata = Field(default_factory=TranscriptionMetadata) raw_api_response: dict[str, JsonValue] | None = None + request_manifest: RequestManifest | None = None + transport_evidence: TransportEvidence | None = None @property def finish_reason(self) -> str | None: @@ -91,6 +110,7 @@ class TranscriptionProvider(Protocol): mime_type: str, temperature: float | None = None, top_p: float | None = None, + source_reference: SourceEvidenceReference | None = None, ) -> TranscriptionResult: """Transcribe the provided image according to the prompt text.""" ... diff --git a/src/transcription/providers/evidence.py b/src/transcription/providers/evidence.py new file mode 100644 index 0000000..08df99d --- /dev/null +++ b/src/transcription/providers/evidence.py @@ -0,0 +1,154 @@ +"""Versioned, provider-neutral contracts for processing evidence.""" + +from __future__ import annotations + +import hashlib +import json +import os +import platform +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version +from typing import Any +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic import JsonValue + +REQUEST_MANIFEST_SCHEMA = "transcription.request-manifest" +REQUEST_MANIFEST_VERSION = "1" +SOFTWARE_CONTEXT_SCHEMA = "transcription.software-context" +SOFTWARE_CONTEXT_VERSION = "1" +TRANSPORT_EVIDENCE_SCHEMA = "transcription.transport-evidence" +TRANSPORT_EVIDENCE_VERSION = "1" +CANONICAL_JSON_ALGORITHM = "transcription-canonical-json-v1" + +SAFE_RESPONSE_HEADERS = frozenset( + { + "content-type", + "content-encoding", + "date", + "retry-after", + "x-request-id", + "x-openrouter-generation-id", + "x-ratelimit-limit", + "x-ratelimit-remaining", + "x-ratelimit-reset", + } +) + + +class EvidenceModel(BaseModel): + """Strict immutable base for persisted evidence contracts.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + +class SourceEvidenceReference(EvidenceModel): + """Secret-safe identity for source content used by one execution.""" + + source_id: UUID + digest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + byte_size: int = Field(ge=0) + media_type: str = Field(min_length=1) + page_number: int = Field(ge=1) + width: int | None = Field(default=None, ge=1) + height: int | None = Field(default=None, ge=1) + derivative_id: UUID | None = None + transformation: str | None = None + + +class SoftwareContext(EvidenceModel): + """Versions needed to interpret a provider execution.""" + + schema_name: Literal["transcription.software-context"] = SOFTWARE_CONTEXT_SCHEMA + schema_version: Literal["1"] = SOFTWARE_CONTEXT_VERSION + application_version: str + application_commit: str | None = None + adapter_name: str + adapter_version: str + client_library: str + client_library_version: str + python_version: str + + +class RequestManifest(EvidenceModel): + """Frozen, secret-safe representation of one concrete provider request.""" + + schema_name: Literal["transcription.request-manifest"] = REQUEST_MANIFEST_SCHEMA + schema_version: Literal["1"] = REQUEST_MANIFEST_VERSION + provider: str = Field(min_length=1) + requested_model: str = Field(min_length=1) + request: dict[str, JsonValue] + source: SourceEvidenceReference + explicitly_supplied_parameters: tuple[str, ...] = () + omitted_optional_parameters: tuple[str, ...] = () + optional_parameter_states: dict[str, Literal["omitted", "null", "value"]] + prompt_content: str = Field(min_length=1) + prompt_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + timeout_seconds: float = Field(gt=0) + retry_policy: str = Field(min_length=1) + software: SoftwareContext + canonicalization: Literal["transcription-canonical-json-v1"] = CANONICAL_JSON_ALGORITHM + + def canonical_bytes(self) -> bytes: + return canonical_json_bytes(self.model_dump(mode="json")) + + def digest(self) -> str: + return hashlib.sha256(self.canonical_bytes()).hexdigest() + + +class TransportEvidence(EvidenceModel): + """Exact response captured at the application/router HTTP boundary.""" + + schema_name: Literal["transcription.transport-evidence"] = TRANSPORT_EVIDENCE_SCHEMA + schema_version: Literal["1"] = TRANSPORT_EVIDENCE_VERSION + response_received: bool + status_code: int | None = Field(default=None, ge=100, le=599) + body: bytes | None = None + safe_headers: dict[str, str] = Field(default_factory=dict) + content_type: str | None = None + content_encoding: str | None = None + request_id: str | None = None + generation_id: str | None = None + + +def canonical_json_bytes(value: Any) -> bytes: + """Serialize JSON deterministically for evidence integrity hashes.""" + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def filter_safe_response_headers(headers: Any) -> dict[str, str]: + """Return only explicitly allowlisted response headers.""" + return { + str(name).lower(): str(value) for name, value in headers.items() if str(name).lower() in SAFE_RESPONSE_HEADERS + } + + +def package_version(package: str) -> str: + """Return an installed package version without failing evidence capture.""" + try: + return version(package) + except PackageNotFoundError: + return "unknown" + + +def build_software_context(*, adapter_name: str, adapter_version: str, client_library: str) -> SoftwareContext: + """Build the runtime software identity for an execution.""" + return SoftwareContext( + application_version=package_version("transcription"), + application_commit=os.environ.get("TRANSCRIPTION_COMMIT") or None, + adapter_name=adapter_name, + adapter_version=adapter_version, + client_library=client_library, + client_library_version=package_version(client_library), + python_version=platform.python_version(), + ) diff --git a/src/transcription/providers/openrouter.py b/src/transcription/providers/openrouter.py index 2d6644c..6250159 100644 --- a/src/transcription/providers/openrouter.py +++ b/src/transcription/providers/openrouter.py @@ -3,12 +3,15 @@ from __future__ import annotations import base64 +import hashlib import logging from typing import Annotated from typing import Any from typing import Literal +import httpx from openrouter import OpenRouter +from openrouter import errors as openrouter_errors from pydantic import BaseModel from pydantic import ConfigDict from pydantic import Field @@ -24,10 +27,38 @@ from transcription.providers.base import ProviderResponseError from transcription.providers.base import ProviderUsage from transcription.providers.base import TranscriptionMetadata from transcription.providers.base import TranscriptionResult +from transcription.providers.evidence import RequestManifest +from transcription.providers.evidence import SourceEvidenceReference +from transcription.providers.evidence import TransportEvidence +from transcription.providers.evidence import build_software_context +from transcription.providers.evidence import filter_safe_response_headers logger = logging.getLogger(__name__) DEFAULT_OPENROUTER_MODEL = "google/gemini-2.5-flash" +OPENROUTER_ADAPTER_VERSION = "1" + + +class _CapturingAsyncClient: + """Delegate SDK HTTP calls while retaining the response before SDK parsing.""" + + def __init__(self, client: httpx.AsyncClient): + self._client = client + self.last_response: httpx.Response | None = None + + async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response: + response = await self._client.send(request, **kwargs) + self.last_response = response + return response + + def build_request(self, *args: Any, **kwargs: Any) -> httpx.Request: + return self._client.build_request(*args, **kwargs) + + async def aclose(self) -> None: + await self._client.aclose() + + def reset(self) -> None: + self.last_response = None class _ProviderModel(BaseModel): @@ -121,16 +152,47 @@ JSON_OBJECT_ADAPTER = TypeAdapter(dict[str, JsonValue]) class OpenRouterTranscriptionProvider: """Adapter that performs image transcription through OpenRouter.""" - def __init__(self, *, settings: Settings | None = None, client: OpenRouter | None = None): + def __init__( + self, + *, + settings: Settings | None = None, + client: OpenRouter | None = None, + async_client: httpx.AsyncClient | None = None, + ): self._settings = settings or get_settings() self._model = self._settings.provider_model or DEFAULT_OPENROUTER_MODEL - self._client = client or OpenRouter(api_key=self._settings.openrouter_api_key.get_secret_value()) + self._capturing_client: _CapturingAsyncClient | None = None + self._current_request_manifest: RequestManifest | None = None + self._current_transport_evidence: TransportEvidence | None = None + if client is None: + self._capturing_client = _CapturingAsyncClient(async_client or httpx.AsyncClient(follow_redirects=True)) + client = OpenRouter( + api_key=self._settings.openrouter_api_key.get_secret_value(), + async_client=self._capturing_client, + ) + self._client = client @property def model(self) -> str: """Return the resolved OpenRouter model slug.""" return self._model + @property + def current_request_manifest(self) -> RequestManifest | None: + return self._current_request_manifest + + @property + def current_transport_evidence(self) -> TransportEvidence | None: + if self._current_transport_evidence is not None: + return self._current_transport_evidence + if self._current_request_manifest is None: + return None + return self._captured_transport_evidence() + + async def aclose(self) -> None: + if self._capturing_client is not None: + await self._capturing_client.aclose() + async def transcribe( self, *, @@ -139,6 +201,7 @@ class OpenRouterTranscriptionProvider: mime_type: str, temperature: float | None = None, top_p: float | None = None, + source_reference: SourceEvidenceReference | None = None, ) -> TranscriptionResult: """Send prompt + image to OpenRouter and return normalized text output.""" request = self._build_request( @@ -148,21 +211,68 @@ class OpenRouterTranscriptionProvider: temperature=temperature, top_p=top_p, ) + manifest = self._build_request_manifest( + request=request, + prompt_text=prompt_text, + source_reference=source_reference, + temperature=temperature, + top_p=top_p, + ) + self._current_request_manifest = manifest + self._current_transport_evidence = None + if self._capturing_client is not None: + self._capturing_client.reset() try: - response = await self._client.chat.send_async(**request.model_dump(mode="json", exclude_none=True)) + response = await self._client.chat.send_async( + **request.model_dump(mode="json", exclude_none=True), + retries=None, + ) except Exception as exc: - message = str(exc).lower() - if "401" in message or "auth" in message or "api key" in message: - raise ProviderAuthError("OpenRouter authentication failed") from exc - raise ProviderError("OpenRouter request failed") from exc + transport = self._captured_transport_evidence() + self._current_transport_evidence = transport + if isinstance(exc, openrouter_errors.UnauthorizedResponseError): + raise ProviderAuthError( + "OpenRouter authentication failed", + request_manifest=manifest, + transport_evidence=transport, + failure_phase="http_response" if transport.response_received else "connection", + ) from exc + failure_phase = ( + "response_validation" + if isinstance(exc, openrouter_errors.ResponseValidationError) + else "http_response" + if transport.response_received + else "connection" + ) + raise ProviderError( + "OpenRouter request failed", + request_manifest=manifest, + transport_evidence=transport, + failure_phase=failure_phase, + ) from exc + transport = self._captured_transport_evidence() + self._current_transport_evidence = transport raw_api_response = self._coerce_raw_response(response) try: validated_response = OpenRouterResponse.model_validate(raw_api_response) except ValidationError as exc: - raise ProviderResponseError("OpenRouter response failed schema validation") from exc + raise ProviderResponseError( + "OpenRouter response failed schema validation", + request_manifest=manifest, + transport_evidence=transport, + failure_phase="response_validation", + ) from exc - text = self._extract_text(validated_response) + try: + text = self._extract_text(validated_response) + except ProviderResponseError as exc: + raise ProviderResponseError( + str(exc), + request_manifest=manifest, + transport_evidence=transport, + failure_phase="response_validation", + ) from exc model = validated_response.model or self.model metadata = self._build_metadata(validated_response) logger.info("OpenRouter transcription completed using model=%s", model) @@ -178,6 +288,85 @@ class OpenRouterTranscriptionProvider: model=model, metadata=metadata, raw_api_response=raw_api_response, + request_manifest=manifest, + transport_evidence=transport, + ) + + def _build_request_manifest( + self, + *, + request: OpenRouterRequest, + prompt_text: str, + source_reference: SourceEvidenceReference | None, + temperature: float | None, + top_p: float | None, + ) -> RequestManifest | None: + if source_reference is None: + return None + request_payload = request.model_dump(mode="json", exclude_none=True) + sanitized_request = self._replace_embedded_media(request_payload, source_reference=source_reference) + explicit = tuple(name for name, value in (("temperature", temperature), ("top_p", top_p)) if value is not None) + omitted = tuple(name for name in ("temperature", "top_p") if name not in explicit) + return RequestManifest( + provider="openrouter", + requested_model=self.model, + request=JSON_OBJECT_ADAPTER.validate_python(sanitized_request), + source=source_reference, + explicitly_supplied_parameters=explicit, + omitted_optional_parameters=omitted, + optional_parameter_states={ + "temperature": "value" if temperature is not None else "omitted", + "top_p": "value" if top_p is not None else "omitted", + }, + prompt_content=prompt_text, + prompt_sha256=hashlib.sha256(prompt_text.encode("utf-8")).hexdigest(), + timeout_seconds=self._settings.worker_provider_timeout_seconds, + retry_policy="application-bounded; sdk-retries=0", + software=build_software_context( + adapter_name="openrouter", + adapter_version=OPENROUTER_ADAPTER_VERSION, + client_library="openrouter", + ), + ) + + def _replace_embedded_media( + self, + value: Any, + *, + source_reference: SourceEvidenceReference, + ) -> Any: + if isinstance(value, str) and value.startswith("data:") and ";base64," in value: + return { + "source_reference": source_reference.model_dump(mode="json"), + "embedded_media_omitted": True, + } + if isinstance(value, dict): + return { + str(key): self._replace_embedded_media(item, source_reference=source_reference) + for key, item in value.items() + } + if isinstance(value, list | tuple): + return [self._replace_embedded_media(item, source_reference=source_reference) for item in value] + return value + + def _captured_transport_evidence(self) -> TransportEvidence: + response = self._capturing_client.last_response if self._capturing_client is not None else None + if response is None: + return TransportEvidence(response_received=False) + headers = filter_safe_response_headers(response.headers) + try: + body = response.content + except httpx.ResponseNotRead: + body = None + return TransportEvidence( + response_received=True, + status_code=response.status_code, + body=body, + safe_headers=headers, + content_type=headers.get("content-type"), + content_encoding=headers.get("content-encoding"), + request_id=headers.get("x-request-id"), + generation_id=headers.get("x-openrouter-generation-id"), ) def _build_metadata(self, response: OpenRouterResponse) -> TranscriptionMetadata: diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index 5a81d21..b9db18e 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -3,10 +3,12 @@ from datetime import UTC from datetime import datetime from uuid import UUID +from sqlalchemy import func from sqlalchemy.orm import selectinload from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession +from ..db.models import ExecutionAttempt from ..db.models import Job from ..db.models import JobSource from ..db.models import JobSourceStatus @@ -214,7 +216,7 @@ class JobService(ServiceBase): async with self._session_scope(session) as _session: query = ( select(Job) - .options(selectinload(Job.job_sources)) # pyright: ignore[reportArgumentType] + .options(selectinload(Job.job_sources)) .where(Job.id == job_id) .execution_options(populate_existing=True) ) @@ -228,6 +230,22 @@ class JobService(ServiceBase): category=ErrorCategory.VALIDATION, suggestion="Wait for processing to complete, or move the job out of processing before deleting.", ) + attempt_count = ( + await _session.exec( + select(func.count()) + .select_from(ExecutionAttempt) + .where(ExecutionAttempt.job_id == job_id) + ) + ).one() + if attempt_count: + raise JobDeleteBlockedError( + "Job delete blocked because immutable execution evidence exists", + category=ErrorCategory.CONFLICT, + suggestion=( + "Retain the Job as processing history. Evidence deletion requires " + "an explicit retention workflow." + ), + ) for job_source in list(job.job_sources): await _session.delete(job_source) @@ -268,8 +286,6 @@ class JobService(ServiceBase): job_source.raw_transcription = None job_source.error_detail = "Cancelled by user" job_source.executed_at = now - if job_source.source is not None: - job_source.source.raw_transcription = None await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job @@ -310,8 +326,6 @@ class JobService(ServiceBase): job_source.raw_transcription = None job_source.error_detail = None job_source.executed_at = now - if job_source.source is not None: - job_source.source.raw_transcription = None job.status = JobStatus.QUEUED job.date_updated = now diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index 94f8df1..f9f315b 100644 --- a/src/transcription/services/sources.py +++ b/src/transcription/services/sources.py @@ -2,8 +2,10 @@ from __future__ import annotations +import base64 import hashlib import logging +import os from collections.abc import Sequence from contextlib import contextmanager from dataclasses import dataclass @@ -11,6 +13,7 @@ from datetime import UTC from datetime import datetime from pathlib import Path from uuid import UUID +from uuid import uuid4 from pydantic import BaseModel from pydantic import ConfigDict @@ -18,26 +21,34 @@ from pydantic import Field from pydantic import JsonValue from pydantic import TypeAdapter from pydantic import ValidationError +from sqlalchemy import func from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.orm import defer from sqlalchemy.orm import selectinload from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from transcription.config import Settings from transcription.config import get_settings +from transcription.db.models import ExecutionAttempt from transcription.db.models import Job from transcription.db.models import JobSource from transcription.db.models import JobSourceStatus +from transcription.db.models import ProcessingArtifact 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 from transcription.providers import ProviderResponseError +from transcription.providers import RequestManifest +from transcription.providers import SourceEvidenceReference from transcription.providers import TranscriptionMetadata from transcription.providers import TranscriptionProvider from transcription.providers import TranscriptionResult +from transcription.providers import TransportEvidence from transcription.providers import get_transcription_provider +from transcription.providers.evidence import canonical_json_bytes from .base import ServiceBase @@ -96,15 +107,28 @@ class SourceNavigation: class SourceService(ServiceBase): """Manage source records, media payloads, revisions, and page execution output.""" - provider: TranscriptionProvider - def __init__( self, session_factory: async_sessionmaker[AsyncSession] | None = None, settings: Settings | None = None, ): super().__init__(session_factory=session_factory, settings=settings) - self.provider = get_transcription_provider(settings=self.settings) + self._provider: TranscriptionProvider | None = None + + @property + def provider(self) -> TranscriptionProvider: + if self._provider is None: + self._provider = get_transcription_provider(settings=self.settings) + return self._provider + + async def aclose(self) -> None: + """Close provider-owned network resources when they were initialized.""" + if self._provider is None: + return + close = getattr(self._provider, "aclose", None) + if close is not None: + await close() + self._provider = None async def create_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: """Create a new source page record in the database.""" @@ -131,6 +155,7 @@ class SourceService(ServiceBase): query = ( select(Source) .options( + selectinload(Source.document), # pyright: ignore[reportArgumentType] selectinload(Source.job_sources).selectinload(JobSource.job), # pyright: ignore[reportArgumentType] ) .where(Source.id == source_id) @@ -146,6 +171,26 @@ class SourceService(ServiceBase): ) return source + async def read_latest_execution_attempt( + self, + *, + job_source_id: UUID, + session: AsyncSession | None = None, + ) -> ExecutionAttempt | None: + """Read only the latest immutable attempt for one compatibility projection.""" + async with self._session_scope(session) as _session: + query = ( + select(ExecutionAttempt) + .options(defer(ExecutionAttempt.transport_body)) # pyright: ignore[reportArgumentType] + .where(ExecutionAttempt.job_source_id == job_source_id) + .order_by( + ExecutionAttempt.attempt_number.desc(), # pyright: ignore[reportAttributeAccessIssue] + ExecutionAttempt.id.desc(), # pyright: ignore[reportAttributeAccessIssue] + ) + .limit(1) + ) + return (await _session.exec(query)).first() + async def read_source_navigation( self, source_id: UUID, @@ -182,13 +227,8 @@ class SourceService(ServiceBase): return merged async def delete_source(self, source: Source, *, session: AsyncSession | None = None) -> None: - """Delete a source page record.""" - source_file_path = source.file_path - async with self._session_scope(session) as _session: - await _session.delete(source) - await self._finalize(session=_session, caller_session=session) - - self._delete_source_file(source_file_path=source_file_path) + """Delete a source only when it has no retained execution evidence.""" + await self.delete_unlinked_source(source_id=source.id, session=session) async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None: """Delete a source only when no JobSource links exist.""" @@ -198,6 +238,7 @@ class SourceService(ServiceBase): source_id, options=( selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] + selectinload(Source.processing_artifacts), # pyright: ignore[reportArgumentType] ), ) if source is None: @@ -207,11 +248,11 @@ class SourceService(ServiceBase): suggestion="Verify the source id and retry.", ) - if source.job_sources: + if source.job_sources or source.processing_artifacts: raise SourceDeleteBlockedError( - "Source delete blocked because it is linked to one or more jobs", + "Source delete blocked because retained execution evidence exists", category=ErrorCategory.VALIDATION, - suggestion="Remove JobSource links first, then retry deletion.", + suggestion="Preserve the source or use an explicit evidence-retention workflow.", ) source_file_path = source.file_path @@ -339,6 +380,7 @@ class SourceService(ServiceBase): source_id, options=( selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] + selectinload(Source.processing_artifacts), # pyright: ignore[reportArgumentType] ), ) if source is None: @@ -349,6 +391,20 @@ class SourceService(ServiceBase): ) linked_job_sources = list(source.job_sources) + attempt_count = ( + await _session.exec( + select(func.count()) + .select_from(ExecutionAttempt) + .where(ExecutionAttempt.source_id == source_id) + ) + ).one() + if source.processing_artifacts or attempt_count: + raise SourceDeleteBlockedError( + "Source delete blocked because immutable evidence exists", + category=ErrorCategory.VALIDATION, + suggestion="Preserve the source or use an explicit evidence-retention workflow.", + ) + matching_links = [job_source for job_source in linked_job_sources if job_source.job_id == job_id] if not matching_links: raise TranscriptionNotFoundError( @@ -415,6 +471,13 @@ class SourceService(ServiceBase): raw_api_response: dict[str, JsonValue] | None = None, provider: str | None = None, model: str | None = None, + request_manifest: RequestManifest | None = None, + transport_evidence: TransportEvidence | None = None, + failure_phase: str | None = None, + error_category: str | None = None, + started_at: datetime | None = None, + finished_at: datetime | None = None, + duration_ms: int | None = None, session: AsyncSession | None = None, ) -> JobSource: """Persist transcription fields for one source within a specific job.""" @@ -474,9 +537,343 @@ class SourceService(ServiceBase): job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED job_source.executed_at = datetime.now(UTC) - await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source)) + finish_time = finished_at or datetime.now(UTC) + start_time = started_at or finish_time + attempt_number = ( + await _session.exec( + select(func.max(ExecutionAttempt.attempt_number)) + .where(ExecutionAttempt.job_id == job_id) + .where(ExecutionAttempt.source_id == source_id) + ) + ).one() + transport = transport_evidence or TransportEvidence(response_received=False) + manifest_payload = request_manifest.model_dump(mode="json") if request_manifest is not None else None + software_payload = ( + request_manifest.software.model_dump(mode="json") if request_manifest is not None else None + ) + attempt = ExecutionAttempt( + job_source_id=job_source.id, + job_id=job_id, + source_id=source_id, + attempt_number=(attempt_number or 0) + 1, + status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, + provider=provider or job.provider or self.settings.provider.value, + model=model or job.model, + request_manifest=manifest_payload, + request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None, + request_manifest_schema_version=( + request_manifest.schema_version if request_manifest is not None else None + ), + response_received=transport.response_received, + transport_status_code=transport.status_code, + transport_body=transport.body, + transport_content_type=transport.content_type, + transport_content_encoding=transport.content_encoding, + transport_safe_headers=transport.safe_headers or None, + router_request_id=transport.request_id, + router_generation_id=transport.generation_id, + sdk_response_snapshot=raw_response_payload, + normalized_metadata=metadata_payload, + software_context=software_payload, + raw_transcription=text, + error_category=error_category, + error_detail=error_detail, + failure_phase=failure_phase, + started_at=start_time, + finished_at=finish_time, + duration_ms=duration_ms + if duration_ms is not None + else max(0, int((finish_time - start_time).total_seconds() * 1000)), + ) + _session.add(attempt) + + await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source, attempt)) return job_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).options(selectinload(ExecutionAttempt.artifacts)) + 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( + ExecutionAttempt.job_id, + ExecutionAttempt.source_id, + ExecutionAttempt.attempt_number, + ExecutionAttempt.id, + ) + return (await _session.exec(query)).all() + + async def create_processing_artifact( + self, + artifact: ProcessingArtifact, + *, + session: AsyncSession | None = None, + ) -> ProcessingArtifact: + """Persist a validated generic processing artifact.""" + if (artifact.inline_payload is None) == (artifact.external_reference is None): + raise TranscriptionError( + "Processing artifact requires exactly one content location", + category=ErrorCategory.VALIDATION, + suggestion="Provide inline JSON or one stable external reference, but not both.", + ) + self._verify_artifact_integrity(artifact) + async with self._session_scope(session) as _session: + _session.add(artifact) + await self._finalize(session=_session, caller_session=session) + return artifact + + async def create_json_artifact( + self, + *, + source_id: UUID, + execution_attempt_id: UUID | None, + artifact_type: str, + schema_name: str, + schema_version: str, + producer: str, + producer_version: str, + payload: dict[str, JsonValue], + coordinate_metadata: dict[str, JsonValue] | None = None, + session: AsyncSession | None = None, + ) -> ProcessingArtifact: + """Store canonical JSON inline or atomically in the constrained artifact root.""" + if coordinate_metadata is not None: + required = {"units", "origin", "width", "height", "transformations"} + missing = required.difference(coordinate_metadata) + if missing: + raise TranscriptionError( + f"Coordinate metadata is missing required fields: {', '.join(sorted(missing))}", + category=ErrorCategory.VALIDATION, + suggestion="Declare units, origin, dimensions, and transformations.", + ) + payload_bytes = canonical_json_bytes(payload) + artifact_id = uuid4() + inline_payload: dict[str, JsonValue] | None = payload + external_reference: str | None = None + external_path: Path | None = None + if len(payload_bytes) > self.settings.artifact_inline_threshold_bytes: + relative_path = Path(str(source_id)) / f"{artifact_id}.json" + external_path = self.settings.artifact_dir / relative_path + self._write_external_artifact(path=external_path, content=payload_bytes) + inline_payload = None + external_reference = relative_path.as_posix() + artifact = ProcessingArtifact( + id=artifact_id, + source_id=source_id, + execution_attempt_id=execution_attempt_id, + artifact_type=artifact_type, + media_type="application/json", + schema_name=schema_name, + schema_version=schema_version, + producer=producer, + producer_version=producer_version, + inline_payload=inline_payload, + external_reference=external_reference, + payload_sha256=hashlib.sha256(payload_bytes).hexdigest(), + byte_size=len(payload_bytes), + coordinate_metadata=coordinate_metadata, + ) + try: + return await self.create_processing_artifact(artifact, session=session) + except Exception: + if external_path is not None: + external_path.unlink(missing_ok=True) + raise + + def _write_external_artifact(self, *, path: Path, content: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = path.with_suffix(f"{path.suffix}.tmp") + try: + with temporary_path.open("wb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + temporary_path.replace(path) + except OSError as exc: + temporary_path.unlink(missing_ok=True) + raise TranscriptionError( + "Failed to persist external processing artifact", + category=ErrorCategory.INFRA_PERSISTENT, + suggestion="Verify artifact storage permissions and available disk space.", + ) from exc + + def _verify_external_artifact(self, artifact: ProcessingArtifact) -> None: + relative_path = Path(artifact.external_reference or "") + if relative_path.is_absolute() or ".." in relative_path.parts: + raise TranscriptionError( + "External artifact reference must stay inside the artifact root", + category=ErrorCategory.VALIDATION, + suggestion="Use a relative application-managed artifact reference.", + ) + artifact_root = self.settings.artifact_dir.resolve() + artifact_path = (artifact_root / relative_path).resolve() + if artifact_root not in artifact_path.parents or not artifact_path.is_file(): + raise TranscriptionError( + "External processing artifact is unavailable", + category=ErrorCategory.NOT_FOUND, + suggestion="Restore the artifact file or remove its pending database record.", + ) + content = artifact_path.read_bytes() + if len(content) != artifact.byte_size or hashlib.sha256(content).hexdigest() != artifact.payload_sha256: + raise TranscriptionError( + "External processing artifact failed integrity verification", + category=ErrorCategory.CONFLICT, + suggestion="Restore the expected artifact bytes before retrying.", + ) + + def _verify_artifact_integrity(self, artifact: ProcessingArtifact) -> None: + if artifact.inline_payload is None: + self._verify_external_artifact(artifact) + return + content = canonical_json_bytes(artifact.inline_payload) + if len(content) != artifact.byte_size or hashlib.sha256(content).hexdigest() != artifact.payload_sha256: + raise TranscriptionError( + "Inline processing artifact failed integrity verification", + category=ErrorCategory.CONFLICT, + suggestion="Recreate the artifact with its canonical payload digest and byte size.", + ) + + async def list_processing_artifacts( + self, + *, + source_id: UUID, + session: AsyncSession | None = None, + ) -> Sequence[ProcessingArtifact]: + """List generic artifacts associated with a Source.""" + async with self._session_scope(session) as _session: + query = ( + select(ProcessingArtifact) + .where(ProcessingArtifact.source_id == source_id) + .order_by(ProcessingArtifact.created_at, ProcessingArtifact.id) + ) + return (await _session.exec(query)).all() + + async def list_processing_artifact_summaries( + self, + *, + source_id: UUID, + limit: int = 100, + session: AsyncSession | None = None, + ) -> Sequence[ProcessingArtifact]: + """List artifact metadata without loading potentially large inline payloads.""" + async with self._session_scope(session) as _session: + query = ( + select(ProcessingArtifact) + .options(defer(ProcessingArtifact.inline_payload)) # pyright: ignore[reportArgumentType] + .where(ProcessingArtifact.source_id == source_id) + .order_by(ProcessingArtifact.created_at, ProcessingArtifact.id) + .limit(limit) + ) + 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 _session.get(Source, source_id) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session)) + artifacts = list(await self.list_processing_artifacts(source_id=source_id, session=_session)) + + for artifact in artifacts: + self._verify_artifact_integrity(artifact) + + artifact_payloads = [ + { + "id": str(artifact.id), + "source_id": str(artifact.source_id), + "execution_attempt_id": (str(artifact.execution_attempt_id) if artifact.execution_attempt_id else None), + "artifact_type": artifact.artifact_type, + "media_type": artifact.media_type, + "schema_name": artifact.schema_name, + "schema_version": artifact.schema_version, + "producer": artifact.producer, + "producer_version": artifact.producer_version, + "inline_payload": artifact.inline_payload, + "external_reference": artifact.external_reference, + "payload_sha256": artifact.payload_sha256, + "byte_size": artifact.byte_size, + "coordinate_metadata": artifact.coordinate_metadata, + "created_at": artifact.created_at.isoformat(), + } + for artifact in artifacts + ] + 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, + "artifacts": artifact_payloads, + } + async def upsert_revision_for_source( self, *, @@ -588,6 +985,7 @@ async def transcribe_document_image( top_p: float | None = None, settings: Settings | None = None, provider: TranscriptionProvider | None = None, + source_reference: SourceEvidenceReference | None = None, ) -> TranscriptionResult: """Transcribe a local image using the configured prompt and provider.""" runtime_settings = settings or get_settings() @@ -605,17 +1003,25 @@ async def transcribe_document_image( ) image_bytes, mime_type = load_source_payload(image_path) + owns_adapter = provider is None adapter = provider or get_transcription_provider(settings=runtime_settings) logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type) - with handle_transcription_errors(): - result = await adapter.transcribe( - prompt_text=prompt_execution.user_prompt, - image_bytes=image_bytes, - mime_type=mime_type, - temperature=prompt_execution.temperature, - top_p=prompt_execution.top_p, - ) + try: + with handle_transcription_errors(): + result = await adapter.transcribe( + prompt_text=prompt_execution.user_prompt, + image_bytes=image_bytes, + mime_type=mime_type, + temperature=prompt_execution.temperature, + top_p=prompt_execution.top_p, + source_reference=source_reference, + ) + finally: + if owns_adapter: + close = getattr(adapter, "aclose", None) + if close is not None: + await close() logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider) return TranscriptionResult( text=result.text, @@ -629,6 +1035,8 @@ async def transcribe_document_image( model=result.model, metadata=result.metadata, raw_api_response=result.raw_api_response, + request_manifest=result.request_manifest, + transport_evidence=result.transport_evidence, ) diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 8abfa14..172973c 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -1,5 +1,9 @@ import asyncio +import inspect import logging +from dataclasses import dataclass +from datetime import UTC +from datetime import datetime from sqlmodel.ext.asyncio.session import AsyncSession @@ -13,16 +17,47 @@ from ..errors import AppError from ..errors import ErrorCategory from ..errors import classify_unexpected_error from ..errors import format_error_detail +from ..providers import ProviderError +from ..providers import RequestManifest +from ..providers import SourceEvidenceReference +from ..providers import TranscriptionProvider from ..providers import TranscriptionResult +from ..providers import TransportEvidence from . import ServiceBundle from .sources import PromptExecution from .sources import build_prompt_execution from .sources import hash_prompt_text +from .sources import source_mime_type from .sources import transcribe_document_image logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class _SuccessfulPage: + source: Source + result: TranscriptionResult + started_at: datetime + finished_at: datetime + duration_ms: int + + +@dataclass(frozen=True) +class _FailedPage: + source: Source + error: AppError + started_at: datetime + finished_at: datetime + duration_ms: int + request_manifest: RequestManifest | None = None + transport_evidence: TransportEvidence | None = None + failure_phase: str | None = None + sdk_response_snapshot: dict | None = None + normalized_metadata: dict | None = None + provider: str | None = None + model: str | None = None + + async def advance_job( job: Job, services: ServiceBundle, @@ -53,7 +88,7 @@ async def advance_job( return -async def process_queued_job( +async def process_queued_job( # noqa: PLR0915 *, job: Job, services: ServiceBundle, @@ -86,8 +121,8 @@ async def process_queued_job( if not sources: return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session) - successful_pages: list[tuple[Source, TranscriptionResult]] = [] - failed_pages: list[tuple[Source, AppError]] = [] + successful_pages: list[_SuccessfulPage] = [] + failed_pages: list[_FailedPage] = [] externally_stopped = False prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings) @@ -97,21 +132,29 @@ async def process_queued_job( externally_stopped = True break - started_at = asyncio.get_running_loop().time() + started_at = datetime.now(UTC) + monotonic_started_at = asyncio.get_running_loop().time() + result: TranscriptionResult | None = None + page_outcome: _SuccessfulPage | _FailedPage try: + source_reference = SourceEvidenceReference( + source_id=source.id, + digest_sha256=source.file_hash.lower(), + byte_size=source.file_size_bytes, + media_type=source_mime_type(source.file_path), + page_number=source.page_number, + ) result = await asyncio.wait_for( - transcribe_document_image( - source.file_path, - prompt_name=prompt_execution.prompt_name, - prompt_text=prompt_execution.user_prompt, - temperature=prompt_execution.temperature, - top_p=prompt_execution.top_p, + _call_transcriber( + source=source, + prompt_execution=prompt_execution, settings=runtime_settings, provider=services.sources.provider, + source_reference=source_reference, ), timeout=runtime_settings.worker_provider_timeout_seconds, ) - elapsed_seconds = asyncio.get_running_loop().time() - started_at + elapsed_seconds = asyncio.get_running_loop().time() - monotonic_started_at logger.info( "Provider response diagnostics operation=worker.provider_response " "job_id=%s document_id=%s source_id=%s provider=%s model=%s " @@ -132,7 +175,15 @@ async def process_queued_job( ) _validate_transcription_quality(result=result, settings=runtime_settings) - successful_pages.append((source, result)) + finished_at = datetime.now(UTC) + page_outcome = _SuccessfulPage( + source=source, + result=result, + started_at=started_at, + finished_at=finished_at, + duration_ms=max(0, int(elapsed_seconds * 1000)), + ) + successful_pages.append(page_outcome) except TimeoutError: error = AppError( f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s", @@ -140,7 +191,25 @@ async def process_queued_job( suggestion="Retry the job. If this repeats, verify provider latency and request payload size.", retriable=True, ) - failed_pages.append((source, error)) + finished_at = datetime.now(UTC) + page_outcome = _FailedPage( + source=source, + error=error, + started_at=started_at, + finished_at=finished_at, + duration_ms=max( + 0, + int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000), + ), + request_manifest=getattr(services.sources.provider, "current_request_manifest", None), + transport_evidence=getattr( + services.sources.provider, + "current_transport_evidence", + None, + ), + failure_phase="local_timeout", + ) + failed_pages.append(page_outcome) logger.error( "Source failed operation=worker.process_job job_id=%s document_id=%s " "source_id=%s error_id=%s category=%s", @@ -157,7 +226,44 @@ async def process_queued_job( case _: error = classify_unexpected_error(exc, operation="worker.process_job") - failed_pages.append((source, error)) + finished_at = datetime.now(UTC) + provider_error = _find_provider_error(exc) + page_outcome = _FailedPage( + source=source, + error=error, + started_at=started_at, + finished_at=finished_at, + duration_ms=max( + 0, + int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000), + ), + request_manifest=( + result.request_manifest + if result is not None + else provider_error.request_manifest + if provider_error is not None + else None + ), + transport_evidence=( + result.transport_evidence + if result is not None + else provider_error.transport_evidence + if provider_error is not None + else None + ), + failure_phase=( + "transcription_quality" + if result is not None + else provider_error.failure_phase + if provider_error is not None + else "application" + ), + sdk_response_snapshot=result.raw_api_response if result is not None else None, + normalized_metadata=result.metadata_payload() if result is not None else None, + provider=result.provider if result is not None else None, + model=result.model if result is not None else None, + ) + failed_pages.append(page_outcome) logger.error( "Source failed operation=worker.process_job job_id=%s document_id=%s " "source_id=%s error_id=%s category=%s", @@ -168,6 +274,13 @@ async def process_queued_job( error.category.value, ) + await _persist_page_outcome_durably( + job=job, + services=services, + page=page_outcome, + session=session, + ) + if await _job_no_longer_processing(job_id=job.id, services=services, session=session): externally_stopped = True break @@ -183,8 +296,6 @@ async def process_queued_job( updated_job = await _finalize_batch_outcome( job=job, services=services, - successful_pages=successful_pages, - failed_pages=failed_pages, status=terminal_status, session=session, ) @@ -258,41 +369,66 @@ async def _finalize_batch_outcome( *, job: Job, services: ServiceBundle, - successful_pages: list[tuple[Source, TranscriptionResult]], - failed_pages: list[tuple[Source, AppError]], status: JobStatus, session: AsyncSession | None = None, ) -> Job: - """Transaction B: write per-source outcomes and terminal job status atomically.""" + """Persist the terminal aggregate status after all page outcomes are durable.""" if session is None: async with services.jobs._session_scope() as local_session: - for source, result in successful_pages: - await services.sources.update_job_source_transcription( - job_id=job.id, - source_id=source.id, - text=result.text, - error_detail=None, - ai_metadata=result.metadata_payload(), - raw_api_response=result.raw_api_response, - provider=result.provider, - model=result.model, - session=local_session, - ) - - for source, error in failed_pages: - await services.sources.update_job_source_transcription( - job_id=job.id, - source_id=source.id, - text=None, - error_detail=format_error_detail(error), - session=local_session, - ) - updated_job = await services.jobs.mark_job_status(job.id, status, session=local_session) await local_session.commit() return updated_job - for source, result in successful_pages: + updated_job = await services.jobs.mark_job_status(job.id, status, session=session) + await session.commit() + return updated_job + + +async def _persist_page_outcome_durably( + *, + job: Job, + services: ServiceBundle, + page: _SuccessfulPage | _FailedPage, + session: AsyncSession | None, +) -> None: + """Commit one completed provider call before processing the next source.""" + task = asyncio.create_task( + _persist_page_outcome(job=job, services=services, page=page, session=session) + ) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + await task + raise + + +async def _persist_page_outcome( + *, + job: Job, + services: ServiceBundle, + page: _SuccessfulPage | _FailedPage, + session: AsyncSession | None, +) -> None: + if session is None: + async with services.sources._session_scope() as local_session: + await _write_page_outcome(job=job, services=services, page=page, session=local_session) + await local_session.commit() + return + + await _write_page_outcome(job=job, services=services, page=page, session=session) + await session.commit() + + +async def _write_page_outcome( + *, + job: Job, + services: ServiceBundle, + page: _SuccessfulPage | _FailedPage, + session: AsyncSession, +) -> None: + if isinstance(page, _SuccessfulPage): + source = page.source + result = page.result await services.sources.update_job_source_transcription( job_id=job.id, source_id=source.id, @@ -302,21 +438,33 @@ async def _finalize_batch_outcome( raw_api_response=result.raw_api_response, provider=result.provider, model=result.model, + request_manifest=result.request_manifest, + transport_evidence=result.transport_evidence, + started_at=page.started_at, + finished_at=page.finished_at, + duration_ms=page.duration_ms, session=session, ) + return - for source, error in failed_pages: - await services.sources.update_job_source_transcription( - job_id=job.id, - source_id=source.id, - text=None, - error_detail=format_error_detail(error), - session=session, - ) - - updated_job = await services.jobs.mark_job_status(job.id, status, session=session) - await session.commit() - return updated_job + await services.sources.update_job_source_transcription( + job_id=job.id, + source_id=page.source.id, + text=None, + error_detail=format_error_detail(page.error), + ai_metadata=page.normalized_metadata, + raw_api_response=page.sdk_response_snapshot, + provider=page.provider, + model=page.model, + request_manifest=page.request_manifest, + transport_evidence=page.transport_evidence, + failure_phase=page.failure_phase, + error_category=page.error.category.value, + started_at=page.started_at, + finished_at=page.finished_at, + duration_ms=page.duration_ms, + session=session, + ) def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None: @@ -379,3 +527,44 @@ def _coerce_job_status(value: object) -> JobStatus | None: return member return None + + +def _find_provider_error(exc: BaseException) -> ProviderError | None: + """Find provider evidence carried through application error translation.""" + current: BaseException | None = exc + while current is not None: + if isinstance(current, ProviderError): + return current + current = current.__cause__ or current.__context__ + return None + + +async def _call_transcriber( + *, + source: Source, + prompt_execution: PromptExecution, + settings: Settings, + provider: TranscriptionProvider, + source_reference: SourceEvidenceReference, +) -> TranscriptionResult: + """Call the current transcriber while supporting legacy injected test doubles.""" + if "source_reference" in inspect.signature(transcribe_document_image).parameters: + return await transcribe_document_image( + source.file_path, + prompt_name=prompt_execution.prompt_name, + prompt_text=prompt_execution.user_prompt, + temperature=prompt_execution.temperature, + top_p=prompt_execution.top_p, + settings=settings, + provider=provider, + source_reference=source_reference, + ) + return await transcribe_document_image( + source.file_path, + prompt_name=prompt_execution.prompt_name, + prompt_text=prompt_execution.user_prompt, + temperature=prompt_execution.temperature, + top_p=prompt_execution.top_p, + settings=settings, + provider=provider, + ) diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index c12a962..535f2c1 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/src/transcription/ui/pages/jobs_page.py @@ -172,9 +172,7 @@ def register_page() -> None: # noqa: PLR0915 render_detail() if job.status in {JobStatus.QUEUED, JobStatus.PROCESSING}: - ui.label("This page updates automatically while the job is active.").classes( - "text-xs ui-text-muted" - ) + ui.label("This page updates automatically while the job is active.").classes("text-xs ui-text-muted") async def refresh_job() -> None: try: @@ -267,7 +265,7 @@ def register_page() -> None: # noqa: PLR0915 metadata_row("Current Status:", job.status.value) metadata_row("Failed Sources:", str(failed_count)) ui.label( - "Resubmit queues only failed linked sources. New results overwrite prior page-level results." + "Resubmit queues only failed linked sources. Prior execution evidence remains preserved." ).classes("text-xs ui-text-muted") async def submit_resubmit() -> None: diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index 433f667..abbd37f 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -2,6 +2,7 @@ from __future__ import annotations +import base64 import json from pathlib import Path from urllib.parse import quote @@ -9,10 +10,13 @@ from uuid import UUID from fastapi import Request from nicegui import ui +from sqlalchemy import inspect as sqlalchemy_inspect from transcription.config import Settings from transcription.config import get_settings +from transcription.db.models import ExecutionAttempt from transcription.db.models import JobSource +from transcription.db.models import ProcessingArtifact from transcription.db.models import Source from transcription.services.sources import SourceDeleteBlockedError from transcription.services.sources import SourceService @@ -115,6 +119,15 @@ def register_page() -> None: # noqa: PLR0915 try: source = await sources_service.read_source_detail(parsed_source_id) navigation = await sources_service.read_source_navigation(parsed_source_id) + latest_job_source = _latest_job_source(source) + latest_attempt = ( + await sources_service.read_latest_execution_attempt(job_source_id=latest_job_source.id) + if latest_job_source is not None + else None + ) + source_artifacts = list( + await sources_service.list_processing_artifact_summaries(source_id=parsed_source_id) + ) except TranscriptionNotFoundError: ui.label("Source not found").classes("text-h6 ui-text-danger p-4") return @@ -122,7 +135,6 @@ def register_page() -> None: # noqa: PLR0915 show_error(exc, title="Load failed", operation="sources.read") return - latest_job_source = _latest_job_source(source) original_transcription = _resolve_original_transcription(source=source, latest_job_source=latest_job_source) with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"): @@ -137,6 +149,14 @@ def register_page() -> None: # noqa: PLR0915 on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back", ).props("flat") + ui.button( + "Export Evidence", + on_click=lambda: _download_evidence( + source_id=source.id, + sources_service=sources_service, + ), + icon="download", + ).props("flat") destructive_button( "Delete Source", on_click=lambda: ui.navigate.to(f"/sources/{source.id}/delete"), @@ -158,7 +178,12 @@ def register_page() -> None: # noqa: PLR0915 latest_job_source=latest_job_source, sources_service=sources_service, ) - _render_source_metadata_column(source=source, latest_job_source=latest_job_source) + _render_source_metadata_column( + source=source, + latest_job_source=latest_job_source, + latest_attempt=latest_attempt, + source_artifacts=source_artifacts, + ) @ui.page("/sources/{source_id}/delete") async def source_delete_page(source_id: str, session_factory: SessionFactoryDep) -> None: @@ -278,10 +303,20 @@ def _render_source_transcription_column( ) -def _render_source_metadata_column(*, source: Source, latest_job_source: JobSource | None) -> None: +def _render_source_metadata_column( + *, + source: Source, + latest_job_source: JobSource | None, + latest_attempt: ExecutionAttempt | None, + source_artifacts: list[ProcessingArtifact], +) -> None: with ui.column().classes("col-span-12 lg:col-span-4 gap-4"): _render_source_metadata_zone(source) - _render_source_job_metadata_zone(latest_job_source) + _render_source_job_metadata_zone( + latest_job_source, + latest_attempt=latest_attempt, + source_artifacts=source_artifacts, + ) _render_source_revision_logistics_zone(source) @@ -295,7 +330,12 @@ def _render_source_metadata_zone(source: Source) -> None: metadata_row("Stored Path:", source.file_path) -def _render_source_job_metadata_zone(latest_job_source: JobSource | None) -> None: +def _render_source_job_metadata_zone( + latest_job_source: JobSource | None, + *, + latest_attempt: ExecutionAttempt | None, + source_artifacts: list[ProcessingArtifact], +) -> None: with archival_card(title="SourceJob Metadata"): if latest_job_source is None: render_empty_state("No job execution metadata available yet.", italic=True) @@ -328,13 +368,96 @@ def _render_source_job_metadata_zone(latest_job_source: JobSource | None) -> Non ui.label("Failure Detail:").classes("ui-text-muted text-xs mb-1") ui.label(latest_job_source.error_detail).classes("p-2 ui-note-box text-xs") - _render_provider_evidence(latest_job_source) + _render_provider_evidence( + latest_job_source, + latest_attempt=latest_attempt, + source_artifacts=source_artifacts, + ) -def _render_provider_evidence(job_source: JobSource) -> None: +def _render_provider_evidence( + job_source: JobSource, + *, + latest_attempt: ExecutionAttempt | None, + source_artifacts: list[ProcessingArtifact], +) -> None: ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3") - _render_json_evidence("AI Metadata", job_source.ai_metadata) - _render_json_evidence("Raw API Response", job_source.raw_api_response) + if latest_attempt is None: + render_empty_state("Exact transport evidence was not captured for this historical execution.", italic=True) + _render_json_evidence("Normalized Metadata (AI Metadata)", job_source.ai_metadata) + _render_json_evidence( + "OpenRouter SDK Response Snapshot (Raw API Response compatibility field)", + job_source.raw_api_response, + ) + _render_json_evidence("Derived Artifacts", _artifact_display(source_artifacts)) + return + + attempt = latest_attempt + metadata_row("Attempt:", str(attempt.attempt_number)) + metadata_row("Duration:", f"{attempt.duration_ms} ms") + _render_json_evidence("Request Manifest", attempt.request_manifest) + _render_json_evidence("Transport Response", _transport_display(attempt)) + _render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot) + _render_json_evidence("Normalized Metadata", attempt.normalized_metadata) + _render_json_evidence("Software Context", attempt.software_context) + _render_json_evidence("Derived Artifacts", _artifact_display(source_artifacts)) + + +def _artifact_display(artifacts: list[ProcessingArtifact]) -> list[dict[str, object]] | None: + payload = [ + { + "id": str(artifact.id), + "type": artifact.artifact_type, + "format": artifact.media_type, + "schema": f"{artifact.schema_name}@{artifact.schema_version}", + "digest_sha256": artifact.payload_sha256, + "coordinate_metadata": artifact.coordinate_metadata, + } + for artifact in artifacts + ] + return payload or None + + +def _transport_display(attempt: ExecutionAttempt) -> dict[str, object]: + body: object | None = None + body_is_deferred = "transport_body" in sqlalchemy_inspect(attempt).unloaded + if body_is_deferred: + body = "Omitted from Source Detail; use Export Evidence to retrieve the exact bytes." + elif attempt.transport_body is not None: + try: + decoded = attempt.transport_body.decode("utf-8") + try: + body = json.loads(decoded) + except json.JSONDecodeError: + body = decoded + except UnicodeDecodeError: + body = { + "encoding": "base64", + "content": base64.b64encode(attempt.transport_body).decode("ascii"), + } + return { + "response_received": attempt.response_received, + "status_code": attempt.transport_status_code, + "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, + "body": body, + } + + +async def _download_evidence(*, source_id: UUID, sources_service: SourceService) -> None: + try: + payload = await sources_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 + ui.download( + json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False).encode("utf-8"), + filename=f"source-{source_id}-evidence-v1.json", + media_type="application/json", + ) def _render_json_evidence(label: str, value: object | None) -> None: diff --git a/src/transcription/worker.py b/src/transcription/worker.py index f6e67f5..ab26a77 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -164,8 +164,11 @@ async def process_next_queued_job( people=PeopleService(session_factory=session_factory), ) - if session is None: - async with session_scope(session_factory=session_factory) as local_session: - return await process_next_queued_job_workflow(services=services, session=local_session) + try: + if session is None: + async with session_scope(session_factory=session_factory) as local_session: + return await process_next_queued_job_workflow(services=services, session=local_session) - return await process_next_queued_job_workflow(services=services, session=session) + return await process_next_queued_job_workflow(services=services, session=session) + finally: + await services.sources.aclose() diff --git a/tests/providers/test_openrouter.py b/tests/providers/test_openrouter.py index 587a219..019bd89 100644 --- a/tests/providers/test_openrouter.py +++ b/tests/providers/test_openrouter.py @@ -1,5 +1,6 @@ """Tests for transcription.providers.openrouter.""" +import asyncio from types import SimpleNamespace import pytest @@ -143,6 +144,21 @@ class TestOpenRouterProviderTranscribe: mime_type="image/png", ) + @pytest.mark.asyncio + async def test_preserves_caller_cancellation(self): + """Caller and shutdown cancellation must not be relabeled as a timeout.""" + provider = OpenRouterTranscriptionProvider( + settings=Settings(openrouter_api_key="test-key"), + client=_FakeClient(error=asyncio.CancelledError()), + ) + + with pytest.raises(asyncio.CancelledError): + await provider.transcribe( + prompt_text="Prompt body", + image_bytes=b"img-bytes", + mime_type="image/png", + ) + @pytest.mark.asyncio async def test_sends_pdf_as_file_content(self): """PDF payloads use OpenRouter's file content contract.""" diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 6bac264..0855257 100644 --- a/tests/services/test_job_service.py +++ b/tests/services/test_job_service.py @@ -365,7 +365,7 @@ class TestJobService: assert failed_entry.status == JobSourceStatus.PENDING assert failed_entry.error_detail is None assert failed_entry.source is not None - assert failed_entry.source.raw_transcription is None + assert failed_entry.source.raw_transcription == "existing text" assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED @pytest.mark.asyncio diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index 20ddd61..cbdffe2 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -1,5 +1,6 @@ """Reliability tests for worker workflow timeout behavior.""" +import asyncio from pathlib import Path from uuid import uuid4 @@ -12,6 +13,7 @@ from transcription.db.models import JobSource from transcription.db.models import JobSourceStatus 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.workflows import process_queued_job @@ -88,3 +90,64 @@ class TestWorkflowReliability: assert result.error_detail is not None assert "timed out" in result.error_detail.lower() assert "20.0s" in result.error_detail + + @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), + ) + async with services.jobs._session_scope() as session: + document = Document(id=uuid4(), name="durability-doc") + session.add(document) + await session.flush() + job = Job(document_id=document.id, status=JobStatus.QUEUED) + session.add(job) + await session.flush() + for page_number in (1, 2): + source = Source( + document_id=document.id, + page_number=page_number, + upload_name=f"page-{page_number}.jpg", + filename=f"page-{page_number}.jpg", + file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")), + file_hash=str(page_number) * 64, + file_size_bytes=1, + ) + session.add(source) + await session.flush() + session.add(JobSource(job_id=job.id, source_id=source.id)) + await session.commit() + loaded = await services.jobs.read_job(job_id=job.id, session=session) + + second_started = asyncio.Event() + release_second = asyncio.Event() + call_count = 0 + + async def _transcribe(image_path, **kwargs): + nonlocal call_count + _ = (image_path, kwargs) + call_count += 1 + if call_count == 2: + second_started.set() + await release_second.wait() + return TranscriptionResult(text=f"page {call_count}", provider="fixture", model="model") + + monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _transcribe) + 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) + assert len(attempts) == 1 + assert attempts[0].raw_transcription == "page 1" + + release_second.set() + result = await task + assert result is not None + assert result.status == JobStatus.TRANSCRIBED diff --git a/tests/test_db.py b/tests/test_db.py index d330b38..baf4998 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -12,6 +12,7 @@ from transcription.db import create_all from transcription.db import dispose_database_runtime from transcription.db import initialize_database_runtime from transcription.db import session_scope +from transcription.db import upgrade_schema from transcription.db.models import DocumentType from transcription.db.models import PersonRole @@ -38,6 +39,8 @@ async def test_create_all_creates_expected_tables(tmp_path): assert "job" in table_names assert "source" in table_names assert "job_source" in table_names + assert "execution_attempt" in table_names + assert "processing_artifact" in table_names assert "revision" not in table_names finally: await dispose_database_runtime() @@ -112,6 +115,55 @@ async def test_create_all_upgrades_existing_person_table_for_family_search(tmp_p await dispose_database_runtime() +@pytest.mark.asyncio +async def test_v42_upgrade_adds_evidence_tables_without_rewriting_legacy_snapshot(tmp_path): + settings = Settings( + openrouter_api_key="test-key", + database=SqliteSettings(path=str(tmp_path / "v42-upgrade.db")), + environment="test", + ) + runtime = initialize_database_runtime(settings=settings) + + try: + async with runtime.engine.begin() as connection: + await connection.execute(text("CREATE TABLE job (id CHAR(32) PRIMARY KEY NOT NULL)")) + await connection.execute(text("CREATE TABLE source (id CHAR(32) PRIMARY KEY NOT NULL)")) + await connection.execute( + text( + "CREATE TABLE job_source (" + "id CHAR(32) PRIMARY KEY NOT NULL, " + "job_id CHAR(32) NOT NULL, " + "source_id CHAR(32) NOT NULL, " + "raw_api_response JSON" + ")" + ) + ) + await connection.execute(text("INSERT INTO job (id) VALUES ('job-1')")) + await connection.execute(text("INSERT INTO source (id) VALUES ('source-1')")) + await connection.execute( + text( + "INSERT INTO job_source (id, job_id, source_id, raw_api_response) " + "VALUES ('link-1', 'job-1', 'source-1', :snapshot)" + ), + {"snapshot": '{"legacy":true}'}, + ) + + await upgrade_schema(engine=runtime.engine) + await upgrade_schema(engine=runtime.engine) + async with runtime.engine.connect() as connection: + table_names = set(await connection.run_sync(lambda c: inspect(c).get_table_names())) + legacy_snapshot = ( + await connection.execute( + text("SELECT raw_api_response FROM job_source WHERE id = 'link-1'") + ) + ).scalar_one() + + assert {"execution_attempt", "processing_artifact"}.issubset(table_names) + assert "legacy" in legacy_snapshot + finally: + await dispose_database_runtime() + + def test_bootstrap_policy_production_defaults_false(): settings = Settings(openrouter_api_key="test-key", environment="production") assert settings.should_bootstrap_schema is False diff --git a/tests/test_v42_evidence.py b/tests/test_v42_evidence.py new file mode 100644 index 0000000..6eedd10 --- /dev/null +++ b/tests/test_v42_evidence.py @@ -0,0 +1,356 @@ +"""Focused V4.2 evidence, integrity, and benchmark tests.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import UTC +from datetime import datetime +from uuid import uuid4 + +import httpx +import pytest + +from transcription.benchmarking import EditorialAssessment +from transcription.benchmarking import score_transcription +from transcription.config import Settings +from transcription.db.models import Document +from transcription.db.models import Job +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus +from transcription.db.models import ProcessingArtifact +from transcription.db.models import Source +from transcription.providers.base import ProviderError +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.jobs import JobDeleteBlockedError +from transcription.services.jobs import JobService +from transcription.services.sources import SourceDeleteBlockedError +from transcription.services.sources import SourceService +from transcription.services.sources import TranscriptionError +from transcription.services.sources import transcribe_document_image + + +@pytest.mark.asyncio +async def test_openrouter_captures_exact_transport_and_secret_safe_manifest(): + response_body = ( + b'{"id":"gen-1","created":1,"model":"vendor/model","object":"chat.completion",' + b'"system_fingerprint":null,"choices":[{"index":0,"finish_reason":"stop",' + b'"message":{"role":"assistant","content":"Transcript"}}],' + b'"unknown_transport_field":{"retained":true}}' + ) + + async def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["authorization"] == "Bearer test-key" + return httpx.Response( + 200, + content=response_body, + headers={ + "Content-Type": "application/json", + "X-Request-Id": "req-123", + "Set-Cookie": "must-not-persist=1", + }, + request=request, + ) + + async_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + provider = OpenRouterTranscriptionProvider( + settings=Settings(openrouter_api_key="test-key"), + async_client=async_client, + ) + result = await provider.transcribe( + prompt_text="Literal prompt", + image_bytes=b"source-bytes", + mime_type="image/png", + source_reference=SourceEvidenceReference( + source_id=uuid4(), + digest_sha256=hashlib.sha256(b"source-bytes").hexdigest(), + byte_size=len(b"source-bytes"), + media_type="image/png", + page_number=1, + ), + ) + + assert result.transport_evidence is not None + assert result.transport_evidence.body == response_body + assert result.transport_evidence.safe_headers == { + "content-type": "application/json", + "x-request-id": "req-123", + } + assert b'"unknown_transport_field":{"retained":true}' in result.transport_evidence.body + assert result.request_manifest is not None + serialized_manifest = json.dumps(result.request_manifest.model_dump(mode="json")) + assert "data:image/png;base64" not in serialized_manifest + assert "test-key" not in serialized_manifest + assert result.request_manifest.omitted_optional_parameters == ("temperature", "top_p") + + +@pytest.mark.asyncio +async def test_openrouter_failure_retains_safe_response_evidence(): + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 500, + content=b'{"error":{"message":"provider unavailable"}}', + headers={"Content-Type": "application/json", "Retry-After": "2", "Set-Cookie": "secret=1"}, + request=request, + ) + + provider = OpenRouterTranscriptionProvider( + settings=Settings(openrouter_api_key="test-key"), + async_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + with pytest.raises(ProviderError) as failure: + await provider.transcribe( + prompt_text="Literal prompt", + image_bytes=b"source-bytes", + mime_type="image/png", + ) + + evidence = failure.value.transport_evidence + assert evidence is not None + assert evidence.status_code == 500 + assert evidence.body == b'{"error":{"message":"provider unavailable"}}' + assert evidence.safe_headers == {"content-type": "application/json", "retry-after": "2"} + + +@pytest.mark.asyncio +async def test_openrouter_does_not_reuse_prior_response_on_connection_failure(): + call_count = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal call_count + call_count += 1 + if call_count == 1: + return httpx.Response(500, content=b'{"error":"first"}', request=request) + raise httpx.ConnectError("connection failed", request=request) + + provider = OpenRouterTranscriptionProvider( + settings=Settings(openrouter_api_key="test-key"), + async_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + with pytest.raises(ProviderError) as first_failure: + await provider.transcribe(prompt_text="First", image_bytes=b"one", mime_type="image/png") + with pytest.raises(ProviderError) as second_failure: + await provider.transcribe(prompt_text="Second", image_bytes=b"two", mime_type="image/png") + + assert first_failure.value.transport_evidence is not None + assert second_failure.value.transport_evidence is not None + assert second_failure.value.transport_evidence.response_received is False + assert second_failure.value.transport_evidence.body is None + await provider.aclose() + + +@pytest.mark.asyncio +async def test_attempts_are_append_only_and_exported_with_integrity(default_session_factory): + documents = DocumentService(session_factory=default_session_factory) + jobs = JobService(session_factory=default_session_factory) + sources = SourceService(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( + Source( + document_id=document.id, + page_number=1, + upload_name="page.png", + filename="page.png", + file_path="page.png", + file_hash="a" * 64, + file_size_bytes=10, + ) + ) + await sources.create_job_source(JobSource(job_id=job.id, source_id=source.id)) + now = datetime.now(UTC) + + await sources.update_job_source_transcription( + job_id=job.id, + source_id=source.id, + text=None, + error_detail="first failed", + error_category="external_provider_error", + failure_phase="connection", + started_at=now, + finished_at=now, + ) + await sources.update_job_source_transcription( + job_id=job.id, + source_id=source.id, + text="second succeeded", + provider="openrouter", + model="vendor/model", + started_at=now, + finished_at=now, + ) + + attempts = await sources.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" + + payload = {"words": [{"text": "second", "polygon": [0, 0, 1, 1]}]} + payload_bytes = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + artifact = await sources.create_processing_artifact( + ProcessingArtifact( + source_id=source.id, + execution_attempt_id=attempts[1].id, + artifact_type="ocr.words", + media_type="application/json", + schema_name="example.ocr.words", + schema_version="1", + producer="fixture", + producer_version="1", + inline_payload=payload, + payload_sha256=hashlib.sha256(payload_bytes).hexdigest(), + byte_size=len(payload_bytes), + coordinate_metadata={ + "units": "normalized", + "origin": "top-left", + "width": 1, + "height": 1, + "transformations": [], + }, + ) + ) + export = await sources.build_evidence_export(source_id=source.id) + assert export["source"]["digest_sha256"] == "a" * 64 + assert [item["attempt_number"] for item in export["attempts"]] == [1, 2] + assert export["artifacts"][0]["id"] == str(artifact.id) + assert "file_path" not in json.dumps(export) + with pytest.raises(JobDeleteBlockedError): + await jobs.delete_job_with_guardrails(job_id=job.id) + + detail = await sources.read_source_detail(source.id) + latest_job_source = detail.latest_job_source + assert latest_job_source is not None + assert latest_job_source.execution_attempts == [] + assert detail.processing_artifacts == [] + latest_attempt = await sources.read_latest_execution_attempt(job_source_id=latest_job_source.id) + assert latest_attempt is not None + assert latest_attempt.attempt_number == 2 + + +def test_benchmark_scoring_preserves_literal_differences(): + score = score_transcription( + execution_attempt_id=uuid4(), + reference="Farm house", + candidate="farm house", + assessment=EditorialAssessment(silent_normalizations=1), + latency_ms=125, + ) + assert score.character_edits == 1 + assert score.word_edits == 1 + assert score.assessment.silent_normalizations == 1 + + +@pytest.mark.asyncio +async def test_large_json_artifact_uses_constrained_atomic_storage( + default_session_factory, + tmp_path, +): + settings = Settings( + openrouter_api_key="test-key", + artifact_dir=tmp_path / "artifacts", + artifact_inline_threshold_bytes=10, + ) + documents = DocumentService(session_factory=default_session_factory, settings=settings) + sources = SourceService(session_factory=default_session_factory, settings=settings) + document = await documents.create_document(Document(name="External Artifact")) + source = await sources.create_source( + Source( + document_id=document.id, + page_number=1, + upload_name="page.png", + filename="page.png", + file_path="page.png", + file_hash="b" * 64, + file_size_bytes=10, + ) + ) + + artifact = await sources.create_json_artifact( + source_id=source.id, + execution_attempt_id=None, + artifact_type="ocr.layout", + schema_name="example.layout", + schema_version="1", + producer="fixture", + producer_version="1", + payload={"blocks": [{"text": "long enough to be external"}]}, + ) + + assert artifact.inline_payload is None + assert artifact.external_reference is not None + stored_path = settings.artifact_dir / artifact.external_reference + assert stored_path.is_file() + assert hashlib.sha256(stored_path.read_bytes()).hexdigest() == artifact.payload_sha256 + with pytest.raises(SourceDeleteBlockedError): + await sources.delete_unlinked_source(source_id=source.id) + + stored_path.write_bytes(b'{"tampered":true}') + with pytest.raises(TranscriptionError, match="integrity verification"): + await sources.build_evidence_export(source_id=source.id) + + +@pytest.mark.asyncio +async def test_rejects_inline_artifact_with_incorrect_integrity(default_session_factory): + documents = DocumentService(session_factory=default_session_factory) + sources = SourceService(session_factory=default_session_factory) + document = await documents.create_document(Document(name="Inline Integrity")) + source = await sources.create_source( + Source( + document_id=document.id, + page_number=1, + upload_name="page.png", + filename="page.png", + file_path="page.png", + file_hash="d" * 64, + file_size_bytes=10, + ) + ) + + with pytest.raises(TranscriptionError, match="integrity verification"): + await sources.create_processing_artifact( + ProcessingArtifact( + source_id=source.id, + artifact_type="ocr.words", + media_type="application/json", + schema_name="example.words", + schema_version="1", + producer="fixture", + producer_version="1", + inline_payload={"words": []}, + payload_sha256="0" * 64, + byte_size=1, + ) + ) + + +@pytest.mark.asyncio +async def test_standalone_transcription_closes_locally_created_provider(tmp_path, monkeypatch): + image_path = tmp_path / "page.png" + image_path.write_bytes(b"image") + + class _Provider: + closed = False + + async def transcribe(self, **kwargs): + _ = kwargs + return TranscriptionResult(text="result", provider="fixture", model="model") + + async def aclose(self): + self.closed = True + + provider = _Provider() + monkeypatch.setattr("transcription.services.sources.get_transcription_provider", lambda **_kwargs: provider) + + result = await transcribe_document_image( + image_path, + prompt_text="Prompt", + settings=Settings(openrouter_api_key="test-key"), + ) + + assert result.text == "result" + assert provider.closed is True diff --git a/tests/test_worker.py b/tests/test_worker.py index 9a6e426..6d5127b 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -3,6 +3,7 @@ import logging import pytest +from transcription.worker import process_next_queued_job from transcription.worker import run_worker_loop @@ -26,4 +27,27 @@ async def test_run_worker_loop_survives_process_next_exception(monkeypatch, capl await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0) assert calls == 2 - assert "Worker loop exception" in caplog.text \ No newline at end of file + assert "Worker loop exception" in caplog.text + + +@pytest.mark.asyncio +async def test_process_next_closes_initialized_provider(monkeypatch): + closed = False + + class _Sources: + async def aclose(self): + nonlocal closed + closed = True + + services = type("_Services", (), {"sources": _Sources()})() + + monkeypatch.setattr("transcription.worker.ServiceBundle", lambda: services) + + async def _no_job(*, services, session): + _ = (services, session) + return False + + monkeypatch.setattr("transcription.worker.process_next_queued_job_workflow", _no_job) + + assert await process_next_queued_job() is False + assert closed is True diff --git a/tests/ui/test_sources_page.py b/tests/ui/test_sources_page.py index 67587fd..2c75988 100644 --- a/tests/ui/test_sources_page.py +++ b/tests/ui/test_sources_page.py @@ -11,6 +11,8 @@ from transcription.db.models import Job from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.db.models import Source +from transcription.providers.evidence import TransportEvidence +from transcription.services.sources import SourceService # --- Unit Tests for Model @property Definitions --- @@ -223,6 +225,48 @@ class TestSourcesPageRendering: assert "finish_reason" in response.text assert "response-123" in response.text + @pytest.mark.asyncio + async def test_source_detail_separates_v42_evidence_layers(self, app_client, seed_job): + app, client = app_client + job_id = await seed_job(filename="evidence-source.png") + async with session_scope(session_factory=app.state.runtime.session_factory) as session: + job = await session.get(Job, job_id) + assert job is not None + source = ( + await session.exec(select(Source).where(Source.document_id == job.document_id)) + ).first() + assert source is not None + source_id = source.id + + service = SourceService(session_factory=app.state.runtime.session_factory) + await service.update_job_source_transcription( + job_id=job_id, + source_id=source_id, + text="V4.2 transcription", + raw_api_response={"id": "sdk-snapshot"}, + ai_metadata={"finish_reason": "stop"}, + provider="openrouter", + model="vendor/model", + transport_evidence=TransportEvidence( + response_received=True, + status_code=200, + body=b'{"id":"transport-response"}', + safe_headers={"content-type": "application/json"}, + content_type="application/json", + ), + ) + + response = client.get(f"/ui/sources/{source_id}") + + assert response.status_code == 200 + assert "Export Evidence" in response.text + assert "Request Manifest" in response.text + assert "Transport Response" in response.text + assert "OpenRouter SDK Response Snapshot" in response.text + assert "Normalized Metadata" in response.text + assert "Software Context" in response.text + assert "Derived Artifacts" in response.text + @pytest.mark.asyncio async def test_source_delete_page_blocks_when_source_is_job_linked( self, app_client, seed_job