V4.2 Updated what ai_raw_response data is being captured. The changes were more extensive than I expected.

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.

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).

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).

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.

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.

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 |

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.

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.

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]

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)

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")

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)

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",
 ]

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."""
         ...

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(),
+    )

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
 
+        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:

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

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,9 +1003,11 @@ 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)
 
+    try:
         with handle_transcription_errors():
             result = await adapter.transcribe(
                 prompt_text=prompt_execution.user_prompt,
@@ -615,7 +1015,13 @@ async def transcribe_document_image(
                 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,
     )
 
 

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,22 +438,34 @@ 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,
+        source_id=page.source.id,
         text=None,
-            error_detail=format_error_detail(error),
+        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,
     )
 
-    updated_job = await services.jobs.mark_job_status(job.id, status, session=session)
-    await session.commit()
-    return updated_job
-
 
 def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None:
     text_chars = len(result.text)
@@ -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,
+    )

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:

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:

src/transcription/worker.py

@@ -164,8 +164,11 @@ async def process_next_queued_job(
             people=PeopleService(session_factory=session_factory),
         )
 
+    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)
+    finally:
+        await services.sources.aclose()

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."""

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

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

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

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

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
 
 
@@ -27,3 +28,26 @@ async def test_run_worker_loop_survives_process_next_exception(monkeypatch, capl
 
     assert calls == 2
     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

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