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

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
 
-        text = self._extract_text(validated_response)
+        try:
+            text = self._extract_text(validated_response)
+        except ProviderResponseError as exc:
+            raise ProviderResponseError(
+                str(exc),
+                request_manifest=manifest,
+                transport_evidence=transport,
+                failure_phase="response_validation",
+            ) from exc
         model = validated_response.model or self.model
         metadata = self._build_metadata(validated_response)
         logger.info("OpenRouter transcription completed using model=%s", model)
@@ -178,6 +288,85 @@ class OpenRouterTranscriptionProvider:
             model=model,
             metadata=metadata,
             raw_api_response=raw_api_response,
+            request_manifest=manifest,
+            transport_evidence=transport,
+        )
+
+    def _build_request_manifest(
+        self,
+        *,
+        request: OpenRouterRequest,
+        prompt_text: str,
+        source_reference: SourceEvidenceReference | None,
+        temperature: float | None,
+        top_p: float | None,
+    ) -> RequestManifest | None:
+        if source_reference is None:
+            return None
+        request_payload = request.model_dump(mode="json", exclude_none=True)
+        sanitized_request = self._replace_embedded_media(request_payload, source_reference=source_reference)
+        explicit = tuple(name for name, value in (("temperature", temperature), ("top_p", top_p)) if value is not None)
+        omitted = tuple(name for name in ("temperature", "top_p") if name not in explicit)
+        return RequestManifest(
+            provider="openrouter",
+            requested_model=self.model,
+            request=JSON_OBJECT_ADAPTER.validate_python(sanitized_request),
+            source=source_reference,
+            explicitly_supplied_parameters=explicit,
+            omitted_optional_parameters=omitted,
+            optional_parameter_states={
+                "temperature": "value" if temperature is not None else "omitted",
+                "top_p": "value" if top_p is not None else "omitted",
+            },
+            prompt_content=prompt_text,
+            prompt_sha256=hashlib.sha256(prompt_text.encode("utf-8")).hexdigest(),
+            timeout_seconds=self._settings.worker_provider_timeout_seconds,
+            retry_policy="application-bounded; sdk-retries=0",
+            software=build_software_context(
+                adapter_name="openrouter",
+                adapter_version=OPENROUTER_ADAPTER_VERSION,
+                client_library="openrouter",
+            ),
+        )
+
+    def _replace_embedded_media(
+        self,
+        value: Any,
+        *,
+        source_reference: SourceEvidenceReference,
+    ) -> Any:
+        if isinstance(value, str) and value.startswith("data:") and ";base64," in value:
+            return {
+                "source_reference": source_reference.model_dump(mode="json"),
+                "embedded_media_omitted": True,
+            }
+        if isinstance(value, dict):
+            return {
+                str(key): self._replace_embedded_media(item, source_reference=source_reference)
+                for key, item in value.items()
+            }
+        if isinstance(value, list | tuple):
+            return [self._replace_embedded_media(item, source_reference=source_reference) for item in value]
+        return value
+
+    def _captured_transport_evidence(self) -> TransportEvidence:
+        response = self._capturing_client.last_response if self._capturing_client is not None else None
+        if response is None:
+            return TransportEvidence(response_received=False)
+        headers = filter_safe_response_headers(response.headers)
+        try:
+            body = response.content
+        except httpx.ResponseNotRead:
+            body = None
+        return TransportEvidence(
+            response_received=True,
+            status_code=response.status_code,
+            body=body,
+            safe_headers=headers,
+            content_type=headers.get("content-type"),
+            content_encoding=headers.get("content-encoding"),
+            request_id=headers.get("x-request-id"),
+            generation_id=headers.get("x-openrouter-generation-id"),
         )
 
     def _build_metadata(self, response: OpenRouterResponse) -> TranscriptionMetadata:

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,17 +1003,25 @@ async def transcribe_document_image(
         )
     image_bytes, mime_type = load_source_payload(image_path)
 
+    owns_adapter = provider is None
     adapter = provider or get_transcription_provider(settings=runtime_settings)
     logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
 
-    with handle_transcription_errors():
-        result = await adapter.transcribe(
-            prompt_text=prompt_execution.user_prompt,
-            image_bytes=image_bytes,
-            mime_type=mime_type,
-            temperature=prompt_execution.temperature,
-            top_p=prompt_execution.top_p,
-        )
+    try:
+        with handle_transcription_errors():
+            result = await adapter.transcribe(
+                prompt_text=prompt_execution.user_prompt,
+                image_bytes=image_bytes,
+                mime_type=mime_type,
+                temperature=prompt_execution.temperature,
+                top_p=prompt_execution.top_p,
+                source_reference=source_reference,
+            )
+    finally:
+        if owns_adapter:
+            close = getattr(adapter, "aclose", None)
+            if close is not None:
+                await close()
     logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
     return TranscriptionResult(
         text=result.text,
@@ -629,6 +1035,8 @@ async def transcribe_document_image(
         model=result.model,
         metadata=result.metadata,
         raw_api_response=result.raw_api_response,
+        request_manifest=result.request_manifest,
+        transport_evidence=result.transport_evidence,
     )
 
 

src/transcription/services/workflows.py

@@ -1,5 +1,9 @@
 import asyncio
+import inspect
 import logging
+from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
 
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -13,16 +17,47 @@ from ..errors import AppError
 from ..errors import ErrorCategory
 from ..errors import classify_unexpected_error
 from ..errors import format_error_detail
+from ..providers import ProviderError
+from ..providers import RequestManifest
+from ..providers import SourceEvidenceReference
+from ..providers import TranscriptionProvider
 from ..providers import TranscriptionResult
+from ..providers import TransportEvidence
 from . import ServiceBundle
 from .sources import PromptExecution
 from .sources import build_prompt_execution
 from .sources import hash_prompt_text
+from .sources import source_mime_type
 from .sources import transcribe_document_image
 
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class _SuccessfulPage:
+    source: Source
+    result: TranscriptionResult
+    started_at: datetime
+    finished_at: datetime
+    duration_ms: int
+
+
+@dataclass(frozen=True)
+class _FailedPage:
+    source: Source
+    error: AppError
+    started_at: datetime
+    finished_at: datetime
+    duration_ms: int
+    request_manifest: RequestManifest | None = None
+    transport_evidence: TransportEvidence | None = None
+    failure_phase: str | None = None
+    sdk_response_snapshot: dict | None = None
+    normalized_metadata: dict | None = None
+    provider: str | None = None
+    model: str | None = None
+
+
 async def advance_job(
     job: Job,
     services: ServiceBundle,
@@ -53,7 +88,7 @@ async def advance_job(
             return
 
 
-async def process_queued_job(
+async def process_queued_job(  # noqa: PLR0915
     *,
     job: Job,
     services: ServiceBundle,
@@ -86,8 +121,8 @@ async def process_queued_job(
     if not sources:
         return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session)
 
-    successful_pages: list[tuple[Source, TranscriptionResult]] = []
-    failed_pages: list[tuple[Source, AppError]] = []
+    successful_pages: list[_SuccessfulPage] = []
+    failed_pages: list[_FailedPage] = []
     externally_stopped = False
 
     prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings)
@@ -97,21 +132,29 @@ async def process_queued_job(
             externally_stopped = True
             break
 
-        started_at = asyncio.get_running_loop().time()
+        started_at = datetime.now(UTC)
+        monotonic_started_at = asyncio.get_running_loop().time()
+        result: TranscriptionResult | None = None
+        page_outcome: _SuccessfulPage | _FailedPage
         try:
+            source_reference = SourceEvidenceReference(
+                source_id=source.id,
+                digest_sha256=source.file_hash.lower(),
+                byte_size=source.file_size_bytes,
+                media_type=source_mime_type(source.file_path),
+                page_number=source.page_number,
+            )
             result = await asyncio.wait_for(
-                transcribe_document_image(
-                    source.file_path,
-                    prompt_name=prompt_execution.prompt_name,
-                    prompt_text=prompt_execution.user_prompt,
-                    temperature=prompt_execution.temperature,
-                    top_p=prompt_execution.top_p,
+                _call_transcriber(
+                    source=source,
+                    prompt_execution=prompt_execution,
                     settings=runtime_settings,
                     provider=services.sources.provider,
+                    source_reference=source_reference,
                 ),
                 timeout=runtime_settings.worker_provider_timeout_seconds,
             )
-            elapsed_seconds = asyncio.get_running_loop().time() - started_at
+            elapsed_seconds = asyncio.get_running_loop().time() - monotonic_started_at
             logger.info(
                 "Provider response diagnostics operation=worker.provider_response "
                 "job_id=%s document_id=%s source_id=%s provider=%s model=%s "
@@ -132,7 +175,15 @@ async def process_queued_job(
             )
 
             _validate_transcription_quality(result=result, settings=runtime_settings)
-            successful_pages.append((source, result))
+            finished_at = datetime.now(UTC)
+            page_outcome = _SuccessfulPage(
+                source=source,
+                result=result,
+                started_at=started_at,
+                finished_at=finished_at,
+                duration_ms=max(0, int(elapsed_seconds * 1000)),
+            )
+            successful_pages.append(page_outcome)
         except TimeoutError:
             error = AppError(
                 f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s",
@@ -140,7 +191,25 @@ async def process_queued_job(
                 suggestion="Retry the job. If this repeats, verify provider latency and request payload size.",
                 retriable=True,
             )
-            failed_pages.append((source, error))
+            finished_at = datetime.now(UTC)
+            page_outcome = _FailedPage(
+                source=source,
+                error=error,
+                started_at=started_at,
+                finished_at=finished_at,
+                duration_ms=max(
+                    0,
+                    int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
+                ),
+                request_manifest=getattr(services.sources.provider, "current_request_manifest", None),
+                transport_evidence=getattr(
+                    services.sources.provider,
+                    "current_transport_evidence",
+                    None,
+                ),
+                failure_phase="local_timeout",
+            )
+            failed_pages.append(page_outcome)
             logger.error(
                 "Source failed operation=worker.process_job job_id=%s document_id=%s "
                 "source_id=%s error_id=%s category=%s",
@@ -157,7 +226,44 @@ async def process_queued_job(
                 case _:
                     error = classify_unexpected_error(exc, operation="worker.process_job")
 
-            failed_pages.append((source, error))
+            finished_at = datetime.now(UTC)
+            provider_error = _find_provider_error(exc)
+            page_outcome = _FailedPage(
+                source=source,
+                error=error,
+                started_at=started_at,
+                finished_at=finished_at,
+                duration_ms=max(
+                    0,
+                    int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
+                ),
+                request_manifest=(
+                    result.request_manifest
+                    if result is not None
+                    else provider_error.request_manifest
+                    if provider_error is not None
+                    else None
+                ),
+                transport_evidence=(
+                    result.transport_evidence
+                    if result is not None
+                    else provider_error.transport_evidence
+                    if provider_error is not None
+                    else None
+                ),
+                failure_phase=(
+                    "transcription_quality"
+                    if result is not None
+                    else provider_error.failure_phase
+                    if provider_error is not None
+                    else "application"
+                ),
+                sdk_response_snapshot=result.raw_api_response if result is not None else None,
+                normalized_metadata=result.metadata_payload() if result is not None else None,
+                provider=result.provider if result is not None else None,
+                model=result.model if result is not None else None,
+            )
+            failed_pages.append(page_outcome)
             logger.error(
                 "Source failed operation=worker.process_job job_id=%s document_id=%s "
                 "source_id=%s error_id=%s category=%s",
@@ -168,6 +274,13 @@ async def process_queued_job(
                 error.category.value,
             )
 
+        await _persist_page_outcome_durably(
+            job=job,
+            services=services,
+            page=page_outcome,
+            session=session,
+        )
+
         if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
             externally_stopped = True
             break
@@ -183,8 +296,6 @@ async def process_queued_job(
     updated_job = await _finalize_batch_outcome(
         job=job,
         services=services,
-        successful_pages=successful_pages,
-        failed_pages=failed_pages,
         status=terminal_status,
         session=session,
     )
@@ -258,41 +369,66 @@ async def _finalize_batch_outcome(
     *,
     job: Job,
     services: ServiceBundle,
-    successful_pages: list[tuple[Source, TranscriptionResult]],
-    failed_pages: list[tuple[Source, AppError]],
     status: JobStatus,
     session: AsyncSession | None = None,
 ) -> Job:
-    """Transaction B: write per-source outcomes and terminal job status atomically."""
+    """Persist the terminal aggregate status after all page outcomes are durable."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            for source, result in successful_pages:
-                await services.sources.update_job_source_transcription(
-                    job_id=job.id,
-                    source_id=source.id,
-                    text=result.text,
-                    error_detail=None,
-                    ai_metadata=result.metadata_payload(),
-                    raw_api_response=result.raw_api_response,
-                    provider=result.provider,
-                    model=result.model,
-                    session=local_session,
-                )
-
-            for source, error in failed_pages:
-                await services.sources.update_job_source_transcription(
-                    job_id=job.id,
-                    source_id=source.id,
-                    text=None,
-                    error_detail=format_error_detail(error),
-                    session=local_session,
-                )
-
             updated_job = await services.jobs.mark_job_status(job.id, status, session=local_session)
             await local_session.commit()
             return updated_job
 
-    for source, result in successful_pages:
+    updated_job = await services.jobs.mark_job_status(job.id, status, session=session)
+    await session.commit()
+    return updated_job
+
+
+async def _persist_page_outcome_durably(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    page: _SuccessfulPage | _FailedPage,
+    session: AsyncSession | None,
+) -> None:
+    """Commit one completed provider call before processing the next source."""
+    task = asyncio.create_task(
+        _persist_page_outcome(job=job, services=services, page=page, session=session)
+    )
+    try:
+        await asyncio.shield(task)
+    except asyncio.CancelledError:
+        await task
+        raise
+
+
+async def _persist_page_outcome(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    page: _SuccessfulPage | _FailedPage,
+    session: AsyncSession | None,
+) -> None:
+    if session is None:
+        async with services.sources._session_scope() as local_session:
+            await _write_page_outcome(job=job, services=services, page=page, session=local_session)
+            await local_session.commit()
+        return
+
+    await _write_page_outcome(job=job, services=services, page=page, session=session)
+    await session.commit()
+
+
+async def _write_page_outcome(
+    *,
+    job: Job,
+    services: ServiceBundle,
+    page: _SuccessfulPage | _FailedPage,
+    session: AsyncSession,
+) -> None:
+    if isinstance(page, _SuccessfulPage):
+        source = page.source
+        result = page.result
         await services.sources.update_job_source_transcription(
             job_id=job.id,
             source_id=source.id,
@@ -302,21 +438,33 @@ async def _finalize_batch_outcome(
             raw_api_response=result.raw_api_response,
             provider=result.provider,
             model=result.model,
+            request_manifest=result.request_manifest,
+            transport_evidence=result.transport_evidence,
+            started_at=page.started_at,
+            finished_at=page.finished_at,
+            duration_ms=page.duration_ms,
             session=session,
         )
+        return
 
-    for source, error in failed_pages:
-        await services.sources.update_job_source_transcription(
-            job_id=job.id,
-            source_id=source.id,
-            text=None,
-            error_detail=format_error_detail(error),
-            session=session,
-        )
-
-    updated_job = await services.jobs.mark_job_status(job.id, status, session=session)
-    await session.commit()
-    return updated_job
+    await services.sources.update_job_source_transcription(
+        job_id=job.id,
+        source_id=page.source.id,
+        text=None,
+        error_detail=format_error_detail(page.error),
+        ai_metadata=page.normalized_metadata,
+        raw_api_response=page.sdk_response_snapshot,
+        provider=page.provider,
+        model=page.model,
+        request_manifest=page.request_manifest,
+        transport_evidence=page.transport_evidence,
+        failure_phase=page.failure_phase,
+        error_category=page.error.category.value,
+        started_at=page.started_at,
+        finished_at=page.finished_at,
+        duration_ms=page.duration_ms,
+        session=session,
+    )
 
 
 def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None:
@@ -379,3 +527,44 @@ def _coerce_job_status(value: object) -> JobStatus | None:
                 return member
 
     return None
+
+
+def _find_provider_error(exc: BaseException) -> ProviderError | None:
+    """Find provider evidence carried through application error translation."""
+    current: BaseException | None = exc
+    while current is not None:
+        if isinstance(current, ProviderError):
+            return current
+        current = current.__cause__ or current.__context__
+    return None
+
+
+async def _call_transcriber(
+    *,
+    source: Source,
+    prompt_execution: PromptExecution,
+    settings: Settings,
+    provider: TranscriptionProvider,
+    source_reference: SourceEvidenceReference,
+) -> TranscriptionResult:
+    """Call the current transcriber while supporting legacy injected test doubles."""
+    if "source_reference" in inspect.signature(transcribe_document_image).parameters:
+        return await transcribe_document_image(
+            source.file_path,
+            prompt_name=prompt_execution.prompt_name,
+            prompt_text=prompt_execution.user_prompt,
+            temperature=prompt_execution.temperature,
+            top_p=prompt_execution.top_p,
+            settings=settings,
+            provider=provider,
+            source_reference=source_reference,
+        )
+    return await transcribe_document_image(
+        source.file_path,
+        prompt_name=prompt_execution.prompt_name,
+        prompt_text=prompt_execution.user_prompt,
+        temperature=prompt_execution.temperature,
+        top_p=prompt_execution.top_p,
+        settings=settings,
+        provider=provider,
+    )

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),
         )
 
-    if session is None:
-        async with session_scope(session_factory=session_factory) as local_session:
-            return await process_next_queued_job_workflow(services=services, session=local_session)
+    try:
+        if session is None:
+            async with session_scope(session_factory=session_factory) as local_session:
+                return await process_next_queued_job_workflow(services=services, session=local_session)
 
-    return await process_next_queued_job_workflow(services=services, session=session)
+        return await process_next_queued_job_workflow(services=services, session=session)
+    finally:
+        await services.sources.aclose()