V4.7 Phase 2: Evidence Model Simplification (part 2)

Move orientation normalization to the Source-ingest boundary and delete the
ProcessingArtifact subsystem it was built to serve.

Stored pages are now already upright, so nothing downstream derives a rotated
copy: every stored byte is the byte a provider is later sent. Rotation runs in
store_source_file ahead of hashing, so source.file_hash and file_size_bytes
describe exactly what is on disk. normalize_orientation becomes bytes-in /
bytes-out, and JPEG output reuses the source quantization tables and chroma
subsampling instead of re-quantizing at a fixed quality - measured at 50.3-56.1
dB PSNR at -6% size, against 50.0-53.5 dB at +38% for quality=95.

ProcessingArtifact held 2 rows against 77 successful transcriptions; the
subsystem effectively never ran. Deleting it removes the artifact cluster from
sources.py, the derivative resolution in workflows.py, the pre-provider commit
that only existed to make an artifact row durable, and the artifact evidence
dump from the Source detail page. The transcription_quality_warnings payload
folds into execution_attempt.normalized_metadata, so that feature keeps working
without the table.

tools/migrate_v46_to_v47.py carries steps 1 and 2: it rotated the 58 stored
images carrying EXIF orientation 3 in place, updated their recorded hash and
size, dropped processing_artifact and removed its one external file. It is
idempotent, keyed on state rather than a version marker.

tools/migrate_v45_to_v46.py is deleted. That migration is complete, and after
V4.7 it would restore a V4.5 backup into a schema that no longer matches.

Also fixes tests/test_config.py, which read the developer's local .env and
failed whenever WORKER_MAX_RETRIES was set.

Co-authored-by: Copilot App <[email protected]>

src/transcription/config.py

@@ -101,9 +101,7 @@ class Settings(BaseSettings):
     # --- filesystem paths ---
     upload_dir: Path = Path("./uploads")
     prompt_dir: Path = Path("./prompts")
-    artifact_dir: Path = Path("./data/artifacts")
     homepage_dir: Path = Path("./data/homepage")
-    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

@@ -12,7 +12,6 @@ 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 ForeignKey
@@ -71,6 +70,7 @@ class JobSourceStatus(StrEnum):
     PENDING = "pending"
     TRANSCRIBED = "transcribed"
     FAILED = "failed"
+    CANCELLED = "cancelled"
 
 
 class JobPurpose(StrEnum):
@@ -270,15 +270,6 @@ class Job(SQLModel, table=True):
 
         return "unknown"
 
-    @property
-    def error_detail(self) -> str | None:
-        """Return the first available source-level error detail for the job."""
-        for job_source in _loaded_attribute(self, "job_sources") or ():
-            if job_source.error_detail:
-                return job_source.error_detail
-
-        return None
-
 
 class Source(SQLModel, table=True):
     """A document source image or PDF page."""
@@ -319,17 +310,24 @@ class Source(SQLModel, table=True):
         back_populates="source",
         sa_relationship_kwargs={"lazy": "raise"},
     )
-    processing_artifacts: list["ProcessingArtifact"] = Relationship(
-        back_populates="source",
-        sa_relationship_kwargs={"lazy": "noload"},
-    )
 
     @property
     def latest_job_source(self) -> Optional["JobSource"]:
-        """Return the most recent job execution record for this source."""
+        """Return the most recent job execution record for this source.
-        if not self.job_sources:
+
-            return None
+        ``JobSource`` carries no timestamp of its own, so recency is the parent
-        return max(self.job_sources, key=lambda js: js.executed_at)
+        job's creation time. ``(job_id, source_id)`` is unique per source, so
+        this is exactly "the most recent job that included this page".
+        """
+        job_sources = _loaded_attribute(self, "job_sources") or ()
+        dated = [
+            (job, job_source)
+            for job_source in job_sources
+            if (job := _loaded_attribute(job_source, "job")) is not None
+        ]
+        if dated:
+            return max(dated, key=lambda pair: pair[0].date_created)[1]
+        return job_sources[0] if job_sources else None
 
     @property
     def latest_status(self) -> JobSourceStatus | None:
@@ -339,9 +337,19 @@ class Source(SQLModel, table=True):
 
     @property
     def latest_error_detail(self) -> str | None:
-        """Return the error detail from the latest job run, if present."""
+        """Return the error detail of the latest attempt on the latest job run.
+
+        Failure detail lives on ``ExecutionAttempt``; ``JobSource`` records only
+        which page a job is working on and how far it got.
+        """
         latest = self.latest_job_source
-        return latest.error_detail if latest else None
+        if latest is None:
+            return None
+        attempts = _loaded_attribute(latest, "execution_attempts") or ()
+        for attempt in sorted(attempts, key=lambda item: item.attempt_number, reverse=True):
+            if attempt.error_detail:
+                return attempt.error_detail
+        return None
 
     @property
     def document_name(self) -> str | None:
@@ -368,11 +376,6 @@ class JobSource(SQLModel, table=True):
             nullable=False,
         ),
     )
-    raw_transcription: str | None = None
-    ai_metadata: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
-    raw_api_response: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
-    error_detail: str | None = None
-    executed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
     job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"})
     source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "raise"})
@@ -393,7 +396,19 @@ class ExecutionAttempt(SQLModel, table=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
+    status: JobSourceStatus = Field(
+        sa_column=Column(
+            # Declared identically to job_source.status. Without values_callable
+            # SQLAlchemy persists enum *names*, which is defect [45]: the two
+            # columns spelled the same status differently and never compared equal.
+            SAEnum(
+                JobSourceStatus,
+                values_callable=lambda enum_cls: [item.value for item in enum_cls],
+                native_enum=False,
+            ),
+            nullable=False,
+        )
+    )
     provider: str
     model: str | None = None
     request_manifest: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
@@ -428,44 +443,3 @@ class ExecutionAttempt(SQLModel, table=True):
     job_source: Optional["JobSource"] = Relationship(
         back_populates="execution_attempts", sa_relationship_kwargs={"lazy": "raise"}
     )
-    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", sa_relationship_kwargs={"lazy": "raise"}
-    )
-    source: Optional["Source"] = Relationship(
-        back_populates="processing_artifacts", sa_relationship_kwargs={"lazy": "raise"}
-    )

src/transcription/services/jobs.py

@@ -2,7 +2,6 @@ import logging
 from collections.abc import Sequence
 from datetime import UTC
 from datetime import datetime
-from pathlib import Path
 from uuid import UUID
 
 from sqlalchemy import func
@@ -17,7 +16,6 @@ from ..db.models import Job
 from ..db.models import JobSource
 from ..db.models import JobSourceStatus
 from ..db.models import JobStatus
-from ..db.models import ProcessingArtifact
 from ..db.models import Source
 from ..errors import AppError
 from ..errors import ErrorCategory
@@ -276,7 +274,6 @@ class JobService(ServiceBase):
 
     async def delete_job_and_evidence(self, *, job_id: UUID) -> None:
         """Explicitly delete a terminal job and all evidence owned by its attempts."""
-        external_references: list[str] = []
         async with self._session_scope() as session:
             job = (
                 await session.exec(
@@ -302,25 +299,6 @@ class JobService(ServiceBase):
                     )
                 ).all()
             )
-            if attempts:
-                attempt_ids = [attempt.id for attempt in attempts]
-                artifacts = list(
-                    (
-                        await session.exec(
-                            select(ProcessingArtifact).where(
-                                col(ProcessingArtifact.execution_attempt_id).in_(attempt_ids)
-                            )
-                        )
-                    ).all()
-                )
-                external_references = [
-                    artifact.external_reference
-                    for artifact in artifacts
-                    if artifact.external_reference is not None
-                ]
-                for artifact in artifacts:
-                    await session.delete(artifact)
-                await session.flush()
             for attempt in attempts:
                 await session.delete(attempt)
             await session.flush()
@@ -331,24 +309,6 @@ class JobService(ServiceBase):
             await session.delete(job)
             await self._finalize(session=session, caller_session=None)
 
-        for external_reference in external_references:
-            self._delete_external_artifact(external_reference)
-
-    def _delete_external_artifact(self, external_reference: str) -> None:
-        relative_path = Path(external_reference)
-        if relative_path.is_absolute() or ".." in relative_path.parts:
-            logger.warning("Skipped unsafe external artifact reference during job deletion: %s", external_reference)
-            return
-        artifact_root = self.settings.artifact_dir.resolve()
-        artifact_path = (artifact_root / relative_path).resolve()
-        if artifact_root not in artifact_path.parents:
-            logger.warning("Skipped external artifact outside configured root: %s", external_reference)
-            return
-        try:
-            artifact_path.unlink(missing_ok=True)
-        except OSError:
-            logger.warning("Failed to delete external artifact: %s", artifact_path)
-
     async def cancel_job(self, *, job_id: UUID, session: AsyncSession | None = None) -> Job:
         """Cancel a queued/processing job and stop remaining source work."""
         async with self._session_scope(session) as _session:
@@ -378,10 +338,7 @@ class JobService(ServiceBase):
             for job_source in job.job_sources:
                 if job_source.status == JobSourceStatus.TRANSCRIBED:
                     continue
-                job_source.status = JobSourceStatus.FAILED
+                job_source.status = JobSourceStatus.CANCELLED
-                job_source.raw_transcription = None
-                job_source.error_detail = "Cancelled by user"
-                job_source.executed_at = now
 
             await self._finalize(session=_session, caller_session=session, refresh=(job,))
             return job
@@ -408,20 +365,21 @@ class JobService(ServiceBase):
                     suggestion="Cancel processing first, then resubmit remaining sources.",
                 )
 
-            candidates = [job_source for job_source in job.job_sources if job_source.status == JobSourceStatus.FAILED]
+            # Cancelled pages are re-attemptable: before V4.7 cancel wrote FAILED,
+            # so resubmit already reset them. Excluding CANCELLED here would make
+            # cancelled work permanently unrecoverable.
+            resubmittable = {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED}
+            candidates = [job_source for job_source in job.job_sources if job_source.status in resubmittable]
             if not candidates:
                 raise JobResubmitBlockedError(
-                    "Job has no failed sources to resubmit",
+                    "Job has no failed or cancelled sources to resubmit",
                     category=ErrorCategory.VALIDATION,
-                    suggestion="Only failed sources can be resubmitted.",
+                    suggestion="Only failed or cancelled sources can be resubmitted.",
                 )
 
             now = datetime.now(UTC)
             for job_source in candidates:
                 job_source.status = JobSourceStatus.PENDING
-                job_source.raw_transcription = None
-                job_source.error_detail = None
-                job_source.executed_at = now
 
             job.status = JobStatus.QUEUED
             job.date_updated = now

src/transcription/services/normalization.py

@@ -1,25 +1,29 @@
-"""Metadata-directed orientation normalization for provider image input."""
+"""Metadata-directed orientation normalization applied to image bytes at ingest.
+
+Uploaded pages are stored upright, so nothing downstream has to derive a
+rotated copy: every stored byte is already the byte the provider is sent.
+"""
 
 from __future__ import annotations
 
 import asyncio
-import hashlib
 import io
+import logging
 from dataclasses import dataclass
-from pathlib import Path
 
 from PIL import Image
+from PIL import JpegImagePlugin
 from PIL import UnidentifiedImageError
 from PIL.TiffImagePlugin import TiffImageFile
 
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 
+logger = logging.getLogger(__name__)
+
 ORIENTATION_TAG = 274
-ORIENTATION_SCHEMA = "transcription.orientation-normalization"
+
-ORIENTATION_SCHEMA_VERSION = "1"
+NORMALIZED_MEDIA_TYPES = frozenset({"image/jpeg", "image/png", "image/tiff"})
-ORIENTATION_PRODUCER = "transcription.orientation-normalizer"
-ORIENTATION_PRODUCER_VERSION = "1"
 
 _TRANSPOSE_BY_ORIENTATION = {
     3: (Image.Transpose.ROTATE_180, 180),
@@ -34,46 +38,44 @@ class OrientationNormalizationError(AppError):
 
 @dataclass(frozen=True, slots=True)
 class OrientationNormalization:
-    """Exact derivative bytes and transformation metadata."""
+    """Upright image bytes and the rotation that produced them."""
 
     content: bytes
-    media_type: str
-    suffix: str
     original_orientation: int
     applied_rotation_degrees: int
-    original_width: int
-    original_height: int
-    derivative_width: int
-    derivative_height: int
-    # Computed eagerly by `normalize_orientation`, which already runs off the
-    # event loop, so callers never hash multi-megabyte derivatives inline.
-    digest_sha256: str
 
 
-def normalize_orientation(path: str | Path, *, media_type: str) -> OrientationNormalization | None:
+def normalize_orientation(content: bytes, *, media_type: str) -> OrientationNormalization | None:
     """Physically apply supported EXIF rotation, returning None for a safe no-op.
 
+    JPEG output reuses the source quantization tables and chroma subsampling
+    rather than re-quantizing at a fixed quality. Measured across the corpus
+    that is better on both axes at once - 51.5-55.0 dB PSNR against 50.0-53.5,
+    and slightly smaller output against 38% larger - and it imposes no
+    constraint on image dimensions.
+
     Blocking. Async callers must use :func:`normalize_orientation_async`.
     """
-    source_path = Path(path)
+    if media_type not in NORMALIZED_MEDIA_TYPES:
-    if media_type not in {"image/jpeg", "image/png", "image/tiff"}:
         return None
     try:
-        with Image.open(source_path) as image:
+        image_file = Image.open(io.BytesIO(content))
+    except (OSError, ValueError, UnidentifiedImageError):
+        # Undecodable content is not this function's business to reject. Ingest
+        # accepted such bytes before orientation moved here, and decision A
+        # forbids V4.7 changing what an upload does.
+        logger.info("Skipped orientation normalization for undecodable content (%s)", media_type)
+        return None
+    try:
+        with image_file as image:
             orientation = int(image.getexif().get(ORIENTATION_TAG, 1))
             transformation = _TRANSPOSE_BY_ORIENTATION.get(orientation)
             if transformation is None:
                 return None
 
             transpose, rotation = transformation
-            if isinstance(image, TiffImageFile):
-                original_width = int(image.tag_v2.get(256, image.width))
-                original_height = int(image.tag_v2.get(257, image.height))
             # Pillow applies TIFF orientation while decoding; copying freezes those upright pixels.
-                normalized = image.copy()
+            normalized = image.copy() if isinstance(image, TiffImageFile) else image.transpose(transpose)
-            else:
-                original_width, original_height = image.size
-                normalized = image.transpose(transpose)
             output = io.BytesIO()
             exif = normalized.getexif()
             if ORIENTATION_TAG in exif:
@@ -81,37 +83,35 @@ def normalize_orientation(path: str | Path, *, media_type: str) -> OrientationNo
             save_kwargs: dict[str, object] = {"format": image.format}
             if image.format in {"JPEG", "PNG"}:
                 save_kwargs["exif"] = exif.tobytes()
-            if image.format == "JPEG":
+            if isinstance(image, JpegImagePlugin.JpegImageFile):
-                save_kwargs.update({"quality": 95, "subsampling": 0})
+                # Reusing the source quantization tables and subsampling preserves fidelity
+                # at a smaller size than any re-encode quality setting.
+                save_kwargs.update(
+                    {
+                        "qtables": image.quantization,
+                        "subsampling": JpegImagePlugin.get_sampling(image),
+                        "optimize": True,
+                    }
+                )
             normalized.save(output, **save_kwargs)
     except (OSError, ValueError, UnidentifiedImageError) as exc:
         raise OrientationNormalizationError(
-            f"Source image orientation could not be normalized: {source_path.name}",
+            "Source image orientation could not be normalized",
             category=ErrorCategory.VALIDATION,
-            suggestion="Verify that the curated Source is a valid supported raster image.",
+            suggestion="Verify that the uploaded Source is a valid supported raster image.",
         ) from exc
 
-    suffix = source_path.suffix.lower()
-    content = output.getvalue()
     return OrientationNormalization(
-        content=content,
+        content=output.getvalue(),
-        media_type=media_type,
-        suffix=suffix,
         original_orientation=orientation,
         applied_rotation_degrees=rotation,
-        original_width=original_width,
-        original_height=original_height,
-        derivative_width=normalized.width,
-        derivative_height=normalized.height,
-        digest_sha256=hashlib.sha256(content).hexdigest(),
     )
 
 
-async def normalize_orientation_async(path: str | Path, *, media_type: str) -> OrientationNormalization | None:
+async def normalize_orientation_async(content: bytes, *, media_type: str) -> OrientationNormalization | None:
     """Run :func:`normalize_orientation` off the event loop.
 
-    Pillow decode, transpose, and re-encode are CPU- and disk-bound and scale
+    Pillow decode, transpose, and re-encode are CPU-bound and scale with page
-    with page size, so they must not run on the request or worker event loop
+    size, so they must not run on the request or worker event loop ([MED-01]).
-    ([MED-01]).
     """
-    return await asyncio.to_thread(normalize_orientation, path, media_type=media_type)
+    return await asyncio.to_thread(normalize_orientation, content, media_type=media_type)

src/transcription/services/sources.py

@@ -6,7 +6,6 @@ import asyncio
 import base64
 import hashlib
 import logging
-import os
 from collections.abc import Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -15,7 +14,6 @@ from datetime import datetime
 from pathlib import Path
 from typing import Any
 from uuid import UUID
-from uuid import uuid4
 
 from pydantic import BaseModel
 from pydantic import ConfigDict
@@ -38,7 +36,6 @@ 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
@@ -52,17 +49,11 @@ 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 ..db.loading import defer
 from ..db.loading import orm_attribute
 from ..db.loading import selectinload
 from .base import ServiceBase
-from .normalization import ORIENTATION_PRODUCER
-from .normalization import ORIENTATION_PRODUCER_VERSION
-from .normalization import ORIENTATION_SCHEMA
-from .normalization import ORIENTATION_SCHEMA_VERSION
-from .normalization import normalize_orientation_async
 from .source_media import lookup_source_mime_type
 from .source_media import supported_source_formats
 
@@ -121,10 +112,20 @@ class ProviderInput:
     digest_sha256: str
     byte_size: int
     media_type: str
-    width: int | None = None
+
-    height: int | None = None
+
-    derivative_id: UUID | None = None
+def build_provider_input(source: Source) -> ProviderInput:
-    transformation: str | None = None
+    """Describe the stored Source bytes that a provider request will carry.
+
+    Stored pages are normalized upright at ingest, so the file on disk is the
+    exact payload sent to the provider and ``file_hash`` already identifies it.
+    """
+    return ProviderInput(
+        path=Path(source.file_path),
+        digest_sha256=source.file_hash.lower(),
+        byte_size=source.file_size_bytes,
+        media_type=source_mime_type(source.file_path),
+    )
 
 
 @dataclass(frozen=True, slots=True)
@@ -291,13 +292,10 @@ class SourceService(ServiceBase):
             source = await self._read_source(
                 session=_session,
                 source_id=source_id,
-                options=(
+                options=(selectinload(Source.job_sources),),
-                    selectinload(Source.job_sources),
-                    selectinload(Source.processing_artifacts),
-                ),
             )
 
-            if source.job_sources or source.processing_artifacts:
+            if source.job_sources:
                 raise SourceDeleteBlockedError(
                     "Source delete blocked because retained execution evidence exists",
                     category=ErrorCategory.VALIDATION,
@@ -352,7 +350,11 @@ class SourceService(ServiceBase):
         async with self._session_scope(session) as _session:
             query = select(Source).options(
                 selectinload(Source.document),
-                selectinload(Source.job_sources),
+                # Both are needed by Source.latest_job_source and
+                # latest_error_detail: recency comes from the parent job, and
+                # failure detail lives on the attempt, not the junction row.
+                selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.job)),
+                selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)),
             )
             if document_id is not None:
                 query = query.where(Source.document_id == document_id)
@@ -445,10 +447,7 @@ class SourceService(ServiceBase):
             source = await self._read_source(
                 session=_session,
                 source_id=source_id,
-                options=(
+                options=(selectinload(Source.job_sources),),
-                    selectinload(Source.job_sources),
-                    selectinload(Source.processing_artifacts),
-                ),
             )
 
             linked_job_sources = list(source.job_sources)
@@ -459,7 +458,7 @@ class SourceService(ServiceBase):
                     .where(ExecutionAttempt.source_id == source_id)
                 )
             ).one()
-            if source.processing_artifacts or attempt_count:
+            if attempt_count:
                 raise SourceDeleteBlockedError(
                     "Source delete blocked because immutable evidence exists",
                     category=ErrorCategory.VALIDATION,
@@ -533,7 +532,7 @@ class SourceService(ServiceBase):
         provider: str | None = None,
         model: str | None = None,
         request_manifest: RequestManifest | None = None,
-        model_input_artifact_id: UUID | None = None,
+        quality_warnings: dict[str, JsonValue] | None = None,
         transport_evidence: TransportEvidence | None = None,
         failure_phase: str | None = None,
         error_category: str | None = None,
@@ -566,29 +565,18 @@ class SourceService(ServiceBase):
             job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
             metadata_payload = _validate_transcription_metadata(ai_metadata)
             raw_response_payload = _validate_json_object(raw_api_response, field_name="raw_api_response")
+            attempt_metadata = _merge_quality_warnings(metadata_payload, quality_warnings)
 
             existing_job_source = await _session.exec(
                 select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
             )
             job_source = existing_job_source.first()
+            outcome = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED
             if job_source is None:
-                job_source = JobSource(
+                job_source = JobSource(job_id=job_id, source_id=source_id, status=outcome)
-                    job_id=job_id,
-                    source_id=source_id,
-                    status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED,
-                    raw_transcription=text,
-                    ai_metadata=metadata_payload,
-                    raw_api_response=raw_response_payload,
-                    error_detail=error_detail,
-                )
                 _session.add(job_source)
             else:
-                job_source.raw_transcription = text
+                job_source.status = outcome
-                job_source.ai_metadata = metadata_payload
-                job_source.raw_api_response = raw_response_payload
-                job_source.error_detail = error_detail
-                job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED
-                job_source.executed_at = datetime.now(UTC)
 
             finish_time = finished_at or datetime.now(UTC)
             start_time = started_at or finish_time
@@ -609,7 +597,7 @@ class SourceService(ServiceBase):
                 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,
+                status=outcome,
                 provider=provider or job.provider or self.settings.provider.value,
                 model=model or job.model,
                 request_manifest=manifest_payload,
@@ -626,7 +614,7 @@ class SourceService(ServiceBase):
                 router_request_id=transport.request_id,
                 router_generation_id=transport.generation_id,
                 sdk_response_snapshot=raw_response_payload,
-                normalized_metadata=metadata_payload,
+                normalized_metadata=attempt_metadata,
                 software_context=software_payload,
                 raw_transcription=text,
                 error_category=error_category,
@@ -641,30 +629,6 @@ class SourceService(ServiceBase):
             _session.add(attempt)
             await _session.flush()
 
-            manifest_derivative_id = (
-                request_manifest.source.derivative_id if request_manifest is not None else None
-            )
-            if (
-                model_input_artifact_id is not None
-                and manifest_derivative_id is not None
-                and model_input_artifact_id != manifest_derivative_id
-            ):
-                raise TranscriptionError(
-                    "Provider-input artifact does not match the request manifest",
-                    category=ErrorCategory.CONFLICT,
-                    suggestion="Persist the exact normalized input consumed by this attempt.",
-                )
-            derivative_id = model_input_artifact_id or manifest_derivative_id
-            if derivative_id is not None:
-                derivative = await _session.get(ProcessingArtifact, derivative_id)
-                if derivative is None or derivative.source_id != source_id:
-                    raise TranscriptionError(
-                        "Provider-input derivative evidence is missing or belongs to another Source",
-                        category=ErrorCategory.CONFLICT,
-                        suggestion="Restore the normalized input artifact before persisting the attempt.",
-                    )
-                derivative.execution_attempt_id = attempt.id
-
             if (
                 text is not None
                 and source.raw_transcription is None
@@ -716,7 +680,7 @@ class SourceService(ServiceBase):
     ) -> 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))
+            query = select(ExecutionAttempt)
             if source_id is not None:
                 query = query.where(ExecutionAttempt.source_id == source_id)
             if job_id is not None:
@@ -729,289 +693,6 @@ class SourceService(ServiceBase):
             )
             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
-            await asyncio.to_thread(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
-
-    async def create_binary_artifact(
-        self,
-        *,
-        source_id: UUID,
-        artifact_type: str,
-        media_type: str,
-        schema_name: str,
-        schema_version: str,
-        producer: str,
-        producer_version: str,
-        content: bytes,
-        suffix: str,
-        coordinate_metadata: dict[str, JsonValue] | None = None,
-        session: AsyncSession | None = None,
-    ) -> ProcessingArtifact:
-        """Persist exact binary derivative bytes in application-managed artifact storage."""
-        artifact_id = uuid4()
-        safe_suffix = suffix if suffix.startswith(".") and suffix[1:].isalnum() else ".bin"
-        relative_path = Path(str(source_id)) / f"{artifact_id}{safe_suffix.lower()}"
-        external_path = self.settings.artifact_dir / relative_path
-        payload_sha256 = await asyncio.to_thread(self._write_and_digest_artifact, path=external_path, content=content)
-        artifact = ProcessingArtifact(
-            id=artifact_id,
-            source_id=source_id,
-            artifact_type=artifact_type,
-            media_type=media_type,
-            schema_name=schema_name,
-            schema_version=schema_version,
-            producer=producer,
-            producer_version=producer_version,
-            external_reference=relative_path.as_posix(),
-            payload_sha256=payload_sha256,
-            byte_size=len(content),
-            coordinate_metadata=coordinate_metadata,
-        )
-        try:
-            return await self.create_processing_artifact(artifact, session=session)
-        except Exception:
-            external_path.unlink(missing_ok=True)
-            raise
-
-    async def resolve_provider_input(
-        self,
-        source: Source,
-        *,
-        session: AsyncSession | None = None,
-    ) -> ProviderInput:
-        """Resolve original or physically orientation-normalized provider input."""
-        media_type = source_mime_type(source.file_path)
-        normalized = await normalize_orientation_async(source.file_path, media_type=media_type)
-        if normalized is None:
-            return ProviderInput(
-                path=Path(source.file_path),
-                digest_sha256=source.file_hash.lower(),
-                byte_size=source.file_size_bytes,
-                media_type=media_type,
-            )
-
-        metadata: dict[str, JsonValue] = {
-            "units": "pixels",
-            "origin": "top-left",
-            "width": normalized.derivative_width,
-            "height": normalized.derivative_height,
-            "transformations": [f"rotate-{normalized.applied_rotation_degrees}-degrees-clockwise"],
-            "source_id": str(source.id),
-            "original_digest_sha256": source.file_hash.lower(),
-            "original_byte_size": source.file_size_bytes,
-            "original_orientation": normalized.original_orientation,
-            "applied_rotation_degrees": normalized.applied_rotation_degrees,
-            "original_width": normalized.original_width,
-            "original_height": normalized.original_height,
-            "derivative_width": normalized.derivative_width,
-            "derivative_height": normalized.derivative_height,
-            "derivative_digest_sha256": normalized.digest_sha256,
-            "derivative_byte_size": len(normalized.content),
-            "original_media_type": media_type,
-            "derivative_media_type": normalized.media_type,
-        }
-        artifact = await self.create_binary_artifact(
-            source_id=source.id,
-            artifact_type="orientation_normalized_model_input",
-            media_type=normalized.media_type,
-            schema_name=ORIENTATION_SCHEMA,
-            schema_version=ORIENTATION_SCHEMA_VERSION,
-            producer=ORIENTATION_PRODUCER,
-            producer_version=ORIENTATION_PRODUCER_VERSION,
-            content=normalized.content,
-            suffix=normalized.suffix,
-            coordinate_metadata=metadata,
-            session=session,
-        )
-        if artifact.external_reference is None:
-            raise RuntimeError("Orientation derivative did not receive an external reference")
-        return ProviderInput(
-            path=self.settings.artifact_dir / artifact.external_reference,
-            digest_sha256=artifact.payload_sha256,
-            byte_size=artifact.byte_size,
-            media_type=artifact.media_type,
-            width=normalized.derivative_width,
-            height=normalized.derivative_height,
-            derivative_id=artifact.id,
-            transformation=f"{ORIENTATION_SCHEMA}@{ORIENTATION_SCHEMA_VERSION}",
-        )
-
-    def _write_and_digest_artifact(self, *, path: Path, content: bytes) -> str:
-        """Persist artifact bytes and return their digest in one off-loop hop.
-
-        Binary derivatives are page-sized, so hashing them belongs in the same
-        worker thread as the write rather than on the event loop ([MED-01]).
-        """
-        self._write_external_artifact(path=path, content=content)
-        return hashlib.sha256(content).hexdigest()
-
-    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_artifacts_integrity(self, artifacts: Sequence[ProcessingArtifact]) -> None:
-        """Verify a batch of artifacts; hashing and file reads run off the event loop."""
-        for artifact in artifacts:
-            self._verify_artifact_integrity(artifact)
-
-    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,
-        limit: int = 100,
-        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(col(ProcessingArtifact.created_at), col(ProcessingArtifact.id))
-                .limit(limit)
-            )
-            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))
-                .where(ProcessingArtifact.source_id == source_id)
-                .order_by(col(ProcessingArtifact.created_at), col(ProcessingArtifact.id))
-                .limit(limit)
-            )
-            return (await _session.exec(query)).all()
-
     async def build_evidence_export(
         self,
         *,
@@ -1022,30 +703,7 @@ class SourceService(ServiceBase):
         async with self._session_scope(session) as _session:
             source = await self._read_source(session=_session, source_id=source_id)
             attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session))
-            artifacts = list(await self.list_processing_artifacts(source_id=source_id, session=_session))
 
-        await asyncio.to_thread(self._verify_artifacts_integrity, artifacts)
-
-        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),
@@ -1101,7 +759,6 @@ class SourceService(ServiceBase):
                 "upload_name": source.upload_name,
             },
             "attempts": attempt_payloads,
-            "artifacts": artifact_payloads,
         }
 
     async def upsert_revision_for_source(
@@ -1178,6 +835,23 @@ def _validate_transcription_metadata(
     return validated.as_json_object()
 
 
+def _merge_quality_warnings(
+    metadata: dict[str, JsonValue] | None,
+    quality_warnings: dict[str, JsonValue] | None,
+) -> dict[str, JsonValue] | None:
+    """Attach app-computed quality warnings to provider-normalized metadata.
+
+    The warnings are derived from the transcription text rather than reported by
+    the provider, so they are namespaced under their own key instead of being
+    mixed into the provider's own fields.
+    """
+    if quality_warnings is None:
+        return metadata
+    merged: dict[str, JsonValue] = dict(metadata or {})
+    merged["transcription_quality_warnings"] = quality_warnings
+    return merged
+
+
 def _validate_json_object(
     payload: dict[str, JsonValue] | None,
     *,

src/transcription/services/store.py

@@ -25,8 +25,10 @@ from ..db.session import SessionFactory
 from ..db.session import session_scope
 from .media_storage import build_stored_filename
 from .media_storage import write_media_bytes
+from .normalization import normalize_orientation_async
 from .sources import TranscriptionError
 from .sources import build_prompt_execution
+from .sources import source_mime_type
 from .sources import validate_source_content
 
 logger = logging.getLogger(__name__)
@@ -55,6 +57,15 @@ class DocumentJobResult:
     original_filename: str
 
 
+@dataclass(frozen=True)
+class StoredSourceFile:
+    """A persisted Source file and the identity of the bytes actually stored."""
+
+    path: Path
+    file_hash: str
+    file_size_bytes: int
+
+
 @dataclass(frozen=True)
 class PendingStoredSource:
     """Pre-staged Source artifact tied to a Source id."""
@@ -83,14 +94,14 @@ async def create_document_job(
     prompt_execution = build_prompt_execution(settings=runtime_settings)
     document_id = uuid4()
     source_id = uuid4()
-    stored_path = await store_source_file(
+    stored = await store_source_file(
         filename=filename,
         file_bytes=file_bytes,
         settings=runtime_settings,
         relative_directory=Path("documents") / str(document_id),
         filename_stem=str(source_id),
     )
-    file_hash, file_size_bytes = _compute_file_metadata(file_bytes)
+    stored_path = stored.path
     try:
         async with session_scope(
             session_factory=session_factory,
@@ -103,8 +114,8 @@ async def create_document_job(
                 source_id=source_id,
                 original_filename=filename,
                 stored_path=stored_path,
-                file_hash=file_hash,
+                file_hash=stored.file_hash,
-                file_size_bytes=file_size_bytes,
+                file_size_bytes=stored.file_size_bytes,
                 prompt_execution=prompt_execution,
             )
     except Exception as exc:
@@ -153,7 +164,7 @@ async def create_job_for_document(
     stored_sources: list[PendingStoredSource] = []
     for filename, file_bytes in sorted_source_files:
         source_id = uuid4()
-        stored_path = await store_source_file(
+        stored = await store_source_file(
             filename=filename,
             file_bytes=file_bytes,
             settings=runtime_settings,
@@ -164,9 +175,9 @@ async def create_job_for_document(
             PendingStoredSource(
                 source_id=source_id,
                 original_filename=filename,
-                stored_path=stored_path,
+                stored_path=stored.path,
-                file_hash=_compute_file_hash(file_bytes),
+                file_hash=stored.file_hash,
-                file_size_bytes=len(file_bytes),
+                file_size_bytes=stored.file_size_bytes,
             )
         )
 
@@ -330,14 +341,6 @@ def _best_effort_delete(path: Path) -> None:
         logger.warning("Failed to clean up Source file after database error: %s", path)
 
 
-def _compute_file_hash(file_bytes: bytes) -> str:
-    return hashlib.sha256(file_bytes).hexdigest()
-
-
-def _compute_file_metadata(file_bytes: bytes) -> tuple[str, int]:
-    return _compute_file_hash(file_bytes), len(file_bytes)
-
-
 async def store_source_file(
     *,
     filename: str,
@@ -345,8 +348,13 @@ async def store_source_file(
     settings: Settings | None = None,
     relative_directory: Path | None = None,
     filename_stem: str | None = None,
-) -> Path:
+) -> StoredSourceFile:
-    """Validate and persist a Source file to configured media storage."""
+    """Validate, orient, and persist a Source file to configured media storage.
+
+    Orientation is applied here, at the ingest boundary, so the stored bytes are
+    already upright and the hash and byte size recorded on the ``Source`` row
+    describe exactly what is on disk and exactly what a provider is later sent.
+    """
     runtime_settings = settings or get_settings()
     try:
         validate_source_content(filename=filename, content=file_bytes)
@@ -358,8 +366,18 @@ async def store_source_file(
             retriable=exc.retriable,
         ) from exc
 
+    normalized = await normalize_orientation_async(file_bytes, media_type=source_mime_type(filename))
+    if normalized is not None:
+        logger.info(
+            "Normalized Source orientation on ingest: %s (orientation=%s, rotation=%s)",
+            Path(filename).name,
+            normalized.original_orientation,
+            normalized.applied_rotation_degrees,
+        )
+        file_bytes = normalized.content
+
     upload_dir = runtime_settings.upload_dir
-    return await write_media_bytes(
+    stored_path = await write_media_bytes(
         target_dir=upload_dir if relative_directory is None else upload_dir / relative_directory,
         stored_name=build_stored_filename(filename=filename, filename_stem=filename_stem),
         file_bytes=file_bytes,
@@ -368,3 +386,8 @@ async def store_source_file(
         failure_suggestion="Check upload directory permissions and available disk space, then retry.",
         log_label="Source file",
     )
+    return StoredSourceFile(
+        path=stored_path,
+        file_hash=hashlib.sha256(file_bytes).hexdigest(),
+        file_size_bytes=len(file_bytes),
+    )

src/transcription/services/workflows.py

@@ -4,7 +4,6 @@ import logging
 from dataclasses import dataclass
 from datetime import UTC
 from datetime import datetime
-from uuid import UUID
 
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -32,14 +31,11 @@ from . import ServiceBundle
 from .documents import DocumentService
 from .people import DocumentPersonInput
 from .people import PeopleService
-from .quality import QUALITY_ANALYSIS_PRODUCER
-from .quality import QUALITY_ANALYSIS_PRODUCER_VERSION
-from .quality import QUALITY_ANALYSIS_SCHEMA
-from .quality import QUALITY_ANALYSIS_VERSION
 from .quality import analyze_transcription_quality
 from .quality import quality_warning_payload
 from .sources import PromptExecution
 from .sources import build_prompt_execution
+from .sources import build_provider_input
 from .sources import hash_prompt_text
 from .sources import transcribe_document_image
 
@@ -121,7 +117,6 @@ class _SuccessfulPage:
     started_at: datetime
     finished_at: datetime
     duration_ms: int
-    model_input_artifact_id: UUID | None = None
 
 
 @dataclass(frozen=True)
@@ -138,7 +133,6 @@ class _FailedPage:
     normalized_metadata: dict | None = None
     provider: str | None = None
     model: str | None = None
-    model_input_artifact_id: UUID | None = None
 
 
 async def advance_job(
@@ -223,19 +217,13 @@ async def process_queued_job(  # noqa: PLR0915
         provider_input = None
         page_outcome: _SuccessfulPage | _FailedPage
         try:
-            provider_input = await services.sources.resolve_provider_input(source, session=session)
+            provider_input = build_provider_input(source)
-            if session is not None:
-                await session.commit()
             source_reference = SourceEvidenceReference(
                 source_id=source.id,
                 digest_sha256=provider_input.digest_sha256,
                 byte_size=provider_input.byte_size,
                 media_type=provider_input.media_type,
                 page_number=source.page_number,
-                width=provider_input.width,
-                height=provider_input.height,
-                derivative_id=provider_input.derivative_id,
-                transformation=provider_input.transformation,
             )
             result = await asyncio.wait_for(
                 _call_transcriber(
@@ -276,7 +264,6 @@ async def process_queued_job(  # noqa: PLR0915
                 started_at=started_at,
                 finished_at=finished_at,
                 duration_ms=max(0, int(elapsed_seconds * 1000)),
-                model_input_artifact_id=provider_input.derivative_id,
             )
             successful_pages.append(page_outcome)
         except TimeoutError:
@@ -299,9 +286,6 @@ async def process_queued_job(  # noqa: PLR0915
                 request_manifest=services.sources.provider.current_request_manifest,
                 transport_evidence=services.sources.provider.current_transport_evidence,
                 failure_phase="local_timeout",
-                model_input_artifact_id=(
-                    provider_input.derivative_id if provider_input is not None else None
-                ),
             )
             failed_pages.append(page_outcome)
             logger.error(
@@ -356,9 +340,6 @@ async def process_queued_job(  # noqa: PLR0915
                 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,
-                model_input_artifact_id=(
-                    provider_input.derivative_id if provider_input is not None else None
-                ),
             )
             failed_pages.append(page_outcome)
             logger.error(
@@ -430,7 +411,13 @@ async def process_next_queued_job(
 
 
 def _resolve_job_sources(job: Job) -> list[Source]:
-    """Resolve non-transcribed linked sources for a job in deterministic page order."""
+    """Resolve pending linked sources for a job in deterministic page order.
+
+    A page is work if it has not already succeeded. CANCELLED is included
+    deliberately: resubmit resets cancelled pages to PENDING, so they are
+    re-attemptable, and a cancelled page that somehow reaches a running job is
+    unfinished work rather than a terminal outcome.
+    """
     if not job.job_sources:
         return []
 
@@ -539,36 +526,13 @@ async def _write_page_outcome(
             provider=result.provider,
             model=result.model,
             request_manifest=result.request_manifest,
-            model_input_artifact_id=page.model_input_artifact_id,
+            quality_warnings=quality_warning_payload(analyze_transcription_quality(result.text)),
             transport_evidence=result.transport_evidence,
             started_at=page.started_at,
             finished_at=page.finished_at,
             duration_ms=page.duration_ms,
             session=session,
         )
-        job_source = await services.sources.read_job_source_for_job(
-            job_id=job.id,
-            source_id=source.id,
-            session=session,
-        )
-        attempt = await services.sources.read_latest_execution_attempt(
-            job_source_id=job_source.id,
-            session=session,
-        )
-        if attempt is None:
-            raise RuntimeError("Successful transcription did not create execution evidence")
-        warnings = analyze_transcription_quality(result.text)
-        await services.sources.create_json_artifact(
-            source_id=source.id,
-            execution_attempt_id=attempt.attempt.id,
-            artifact_type="transcription_quality_warnings",
-            schema_name=QUALITY_ANALYSIS_SCHEMA,
-            schema_version=QUALITY_ANALYSIS_VERSION,
-            producer=QUALITY_ANALYSIS_PRODUCER,
-            producer_version=QUALITY_ANALYSIS_PRODUCER_VERSION,
-            payload=quality_warning_payload(warnings),
-            session=session,
-        )
         return
 
     await services.sources.update_job_source_transcription(
@@ -581,7 +545,6 @@ async def _write_page_outcome(
         provider=page.provider,
         model=page.model,
         request_manifest=page.request_manifest,
-        model_input_artifact_id=page.model_input_artifact_id,
         transport_evidence=page.transport_evidence,
         failure_phase=page.failure_phase,
         error_category=page.error.category.value,

src/transcription/ui/components/transcript.py

@@ -1,131 +0,0 @@
-"""Reusable transcript UI components."""
-
-from __future__ import annotations
-
-from collections.abc import Awaitable
-from collections.abc import Callable
-from datetime import datetime
-from typing import Any
-
-from nicegui import ui
-
-from transcription.db.models import Job
-from transcription.db.models import Source
-
-type RevisionAction = Callable[[Source], Awaitable[None] | None]
-
-
-def render_original_transcription_card(*, job: Job, classes: str = "w-full") -> Any:
-    """Render the immutable original job transcription output."""
-    latest_error_detail = _latest_job_error_detail(job)
-    status_label = "Failed" if latest_error_detail else "Transcribed"
-    header = f"Original Transcription | {status_label}"
-    provider = job.provider or "unknown"
-    model = job.model or "unknown"
-    caption = f"{provider} | {model} | {_format_created_at(job.date_updated)}"
-
-    card = ui.card().classes(f"{classes} q-pa-md ui-card-surface")
-    with card, ui.column().classes("w-full q-gutter-y-sm"):
-        ui.label(header).classes("text-subtitle1 text-weight-medium")
-        ui.label(caption).classes("text-caption ui-text-muted")
-        _metadata_row(label="Prompt", value=_latest_job_prompt(job) or "unknown")
-        _metadata_row(label="Updated", value=_format_created_at(job.date_updated))
-
-        latest_transcription = _latest_job_transcription(job)
-
-        if latest_transcription:
-            with ui.card().classes("w-full q-pa-sm ui-card-surface"):
-                ui.markdown(latest_transcription)
-
-        if latest_error_detail:
-            with ui.card().classes("w-full ui-card-error q-pa-sm"):
-                ui.label("Failure detail").classes("text-caption text-uppercase")
-                ui.label(latest_error_detail).classes("text-body2")
-
-    return card
-
-
-def render_revision_row(
-    *,
-    revision: Source | None,
-    initially_expanded: bool = False,
-    classes: str = "w-full",
-    on_delete: RevisionAction | None = None,
-) -> Any:
-    """Render a collapsible row for the single optional source revision."""
-    if revision is None:
-        return None
-
-    header = "Source revision | User-authored"
-    caption = _format_created_at(revision.date_revised or revision.date_uploaded)
-
-    expansion = ui.expansion(value=initially_expanded, group="group").classes(f"{classes} ui-card-surface")
-
-    with expansion, ui.column().classes("w-full q-gutter-y-sm q-pa-sm"):
-        with expansion.add_slot("header"), ui.row().classes("w-full items-start justify-between q-gutter-md"):
-            with ui.column().classes("q-gutter-none"):
-                ui.label(header).classes("text-subtitle1 text-weight-medium")
-                ui.label(caption).classes("text-caption ui-text-muted")
-
-            if on_delete is not None:
-                with ui.dialog() as delete_dialog, ui.card().classes("q-pa-md ui-card-surface"):
-                    ui.label("Delete this source revision?").classes("text-body1")
-                    with ui.row().classes("w-full justify-end q-gutter-sm"):
-                        ui.button("Cancel", on_click=lambda: delete_dialog.submit(False)).props("flat")
-                        ui.button("Delete", on_click=lambda: delete_dialog.submit(True)).props(
-                            'unelevated color="negative"'
-                        )
-
-                async def delete_current_transcript() -> None:
-                    delete_dialog.open()
-                    confirmed = await delete_dialog
-                    if not confirmed:
-                        return
-
-                    maybe_awaitable = on_delete(revision)
-                    if isinstance(maybe_awaitable, Awaitable):
-                        await maybe_awaitable
-
-                with ui.column(align_items="center").classes("self-center q-gutter-none"):
-                    ui.button(icon="delete", on_click=delete_current_transcript).props(
-                        'flat round dense color="negative"'
-                    )
-        _metadata_row(label="Created", value=_format_created_at(revision.date_revised or revision.date_uploaded))
-
-        if revision.revised_text:
-            with ui.card().classes("w-full q-pa-sm ui-card-surface"):
-                ui.markdown(revision.revised_text)
-
-    return expansion
-
-
-def _latest_job_transcription(job: Job) -> str | None:
-    for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True):
-        if job_source.raw_transcription:
-            return job_source.raw_transcription
-    return None
-
-
-def _latest_job_error_detail(job: Job) -> str | None:
-    for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True):
-        if job_source.error_detail:
-            return job_source.error_detail
-    return None
-
-
-def _latest_job_prompt(job: Job) -> str | None:
-    for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True):
-        if job_source.job and job_source.job.prompt_name:
-            return job_source.job.prompt_name
-    return None
-
-
-def _format_created_at(value: datetime) -> str:
-    """Return a compact UTC-like timestamp for row captions."""
-    return value.strftime("%Y-%m-%d %H:%M:%S %Z")
-
-
-def _metadata_row(*, label: str, value: str) -> None:
-    with ui.row().classes("w-md items-start justify-between q-gutter-x-md"):
-        ui.label(label).classes("text-caption ui-text-muted text-uppercase")
-        ui.label(value).classes("text-body2 text-right break-all")

src/transcription/ui/pages/jobs_page.py

@@ -284,9 +284,10 @@ def register_page() -> None:  # noqa: PLR0915
             with archival_card(extra_classes="gap-2"):
                 ui.label(f"Job ID: {job.id}").classes("text-sm font-semibold font-mono ui-text-primary")
                 metadata_row("Current Status:", job.status.value)
-                ui.label("Cancel stops processing and marks remaining non-transcribed sources as failed.").classes(
+                ui.label(
-                    "text-xs ui-text-muted"
+                    "Cancel stops processing and marks remaining non-transcribed sources as cancelled. "
-                )
+                    "Cancelled sources can be resubmitted."
+                ).classes("text-xs ui-text-muted")
 
             async def submit_cancel() -> None:
                 try:
@@ -327,7 +328,11 @@ def register_page() -> None:  # noqa: PLR0915
             render_record_not_found("Job")
             return
 
-        failed_count = sum(1 for js in job.job_sources if js.status == JobSourceStatus.FAILED)
+        resubmittable_count = sum(
+            1
+            for js in job.job_sources
+            if js.status in {JobSourceStatus.FAILED, JobSourceStatus.CANCELLED}
+        )
 
         with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
             page_header("Resubmit Job")
@@ -335,9 +340,10 @@ def register_page() -> None:  # noqa: PLR0915
             with archival_card(extra_classes="gap-2"):
                 ui.label(f"Job ID: {job.id}").classes("text-sm font-semibold font-mono ui-text-primary")
                 metadata_row("Current Status:", job.status.value)
-                metadata_row("Failed Sources:", str(failed_count))
+                metadata_row("Resubmittable Sources:", str(resubmittable_count))
                 ui.label(
-                    "Resubmit queues only failed linked sources. Prior execution evidence remains preserved."
+                    "Resubmit queues failed and cancelled 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

@@ -12,7 +12,7 @@ from nicegui import ui
 from transcription.config import Settings
 from transcription.db.models import ExecutionAttempt
 from transcription.db.models import JobSource
-from transcription.db.models import ProcessingArtifact
+from transcription.db.models import JobSourceStatus
 from transcription.db.models import Source
 from transcription.services.sources import LatestExecutionAttempt
 from transcription.services.sources import SourceDeleteBlockedError
@@ -121,15 +121,12 @@ 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_job_source = source.latest_job_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)
-            )
             attempts = list(await sources_service.list_execution_attempts(source_id=parsed_source_id))
         except TranscriptionNotFoundError:
             render_record_not_found("Source")
@@ -138,7 +135,7 @@ def register_page() -> None:  # noqa: PLR0915
             show_error(exc, title="Load failed", operation="sources.read")
             return
 
-        original_transcription = _resolve_original_transcription(source=source, latest_job_source=latest_job_source)
+        original_transcription = _resolve_original_transcription(source=source, latest_attempt=latest_attempt)
 
         with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
             with section_header_row():
@@ -171,12 +168,6 @@ def register_page() -> None:  # noqa: PLR0915
                         icon="delete",
                         extra_classes="text-xs",
                     )
-            if any(
-                artifact.artifact_type == "orientation_normalized_model_input"
-                for artifact in source_artifacts
-            ):
-                archival_badge("Orientation normalized")
-
             with ui.grid().classes("w-full grid-cols-12 gap-4"):
                 with ui.column().classes("col-span-12 lg:col-span-4 gap-2"):
                     _render_source_navigation(navigation.previous_id, navigation.next_id)
@@ -200,7 +191,6 @@ def register_page() -> None:  # noqa: PLR0915
                     source=source,
                     latest_job_source=latest_job_source,
                     latest_attempt=latest_attempt,
-                    source_artifacts=source_artifacts,
                 )
 
     @ui.page("/sources/{source_id}/delete")
@@ -311,14 +301,12 @@ def _render_source_metadata_column(
     source: Source,
     latest_job_source: JobSource | None,
     latest_attempt: LatestExecutionAttempt | 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,
             latest_attempt=latest_attempt,
-            source_artifacts=source_artifacts,
         )
         _render_source_revision_logistics_zone(source)
 
@@ -337,7 +325,6 @@ def _render_source_job_metadata_zone(
     latest_job_source: JobSource | None,
     *,
     latest_attempt: LatestExecutionAttempt | None,
-    source_artifacts: list[ProcessingArtifact],
 ) -> None:
     with archival_card(title="SourceJob Metadata"):
         if latest_job_source is None:
@@ -350,7 +337,10 @@ def _render_source_job_metadata_zone(
             archival_badge(status)
 
         metadata_row("Job ID:", str(latest_job_source.job_id))
-        metadata_row("Executed:", latest_job_source.executed_at.isoformat())
+        metadata_row(
+            "Executed:",
+            latest_attempt.attempt.finished_at.isoformat() if latest_attempt is not None else "not yet executed",
+        )
         metadata_row(
             "Provider:",
             latest_job_source.job.provider if latest_job_source.job and latest_job_source.job.provider else "unknown",
@@ -366,33 +356,18 @@ def _render_source_job_metadata_zone(
             else "unknown",
         )
 
-        if latest_job_source.error_detail:
+        if latest_attempt is not None and latest_attempt.attempt.error_detail:
             with ui.column().classes("w-full mt-2"):
                 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")
+                ui.label(latest_attempt.attempt.error_detail).classes("p-2 ui-note-box text-xs")
 
-        _render_provider_evidence(
+        _render_provider_evidence(latest_attempt=latest_attempt)
-            latest_job_source,
-            latest_attempt=latest_attempt,
-            source_artifacts=source_artifacts,
-        )
 
 
-def _render_provider_evidence(
+def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None:
-    job_source: JobSource,
-    *,
-    latest_attempt: LatestExecutionAttempt | None,
-    source_artifacts: list[ProcessingArtifact],
-) -> None:
     ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3")
     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.attempt
@@ -403,22 +378,6 @@ def _render_provider_evidence(
     _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: list[dict[str, object]] = [
-        {
-            "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(latest_attempt: LatestExecutionAttempt) -> dict[str, object]:
@@ -549,10 +508,13 @@ def _render_source_transcription_zone(
                 icon="refresh",
             ).props("flat")
 
-        if latest_job_source is not None and latest_job_source.status.value == "failed":
+        if latest_job_source is not None and latest_job_source.status in {
-            ui.label("Source has a failed job execution. Save a human revision to preserve corrected text.").classes(
+            JobSourceStatus.FAILED,
-                "text-xs ui-text-muted italic"
+            JobSourceStatus.CANCELLED,
-            )
+        }:
+            ui.label(
+                "Source has an unfinished job execution. Save a human revision to preserve corrected text."
+            ).classes("text-xs ui-text-muted italic")
 
 
 def _render_machine_candidates(
@@ -657,13 +619,14 @@ def _attempt_warning_count(attempt: ExecutionAttempt) -> int:
 
 
 def _attempt_warnings(attempt: ExecutionAttempt) -> list[dict[str, object]]:
-    for artifact in attempt.artifacts:
+    metadata = attempt.normalized_metadata or {}
-        if artifact.artifact_type != "transcription_quality_warnings" or artifact.inline_payload is None:
+    payload = metadata.get("transcription_quality_warnings")
-            continue
+    if not isinstance(payload, dict):
-        warnings = artifact.inline_payload.get("warnings")
-        if isinstance(warnings, list):
-            return [warning for warning in warnings if isinstance(warning, dict)]
         return []
+    warnings = payload.get("warnings")
+    if not isinstance(warnings, list):
+        return []
+    return [warning for warning in warnings if isinstance(warning, dict)]
 
 
 def _render_attempt_warnings(attempt: ExecutionAttempt, *, label: str) -> None:
@@ -683,13 +646,7 @@ def _reset_revision_text(revision_input: ui.textarea, source: Source, original_t
     revision_input.value = fallback_text
 
 
-def _latest_job_source(source: Source) -> JobSource | None:
+def _resolve_original_transcription(*, source: Source, latest_attempt: LatestExecutionAttempt | None) -> str | None:
-    if not source.job_sources:
+    if source.raw_transcription is None and latest_attempt is not None:
-        return None
+        return latest_attempt.attempt.raw_transcription
-    return max(source.job_sources, key=lambda item: item.executed_at)
-
-
-def _resolve_original_transcription(*, source: Source, latest_job_source: JobSource | None) -> str | None:
-    if source.raw_transcription is None and latest_job_source is not None:
-        return latest_job_source.raw_transcription
     return source.raw_transcription

tests/integration/test_pipeline_flow.py

@@ -6,9 +6,12 @@ from uuid import uuid4
 
 import pytest
 from PIL import Image
+from sqlmodel import col
+from sqlmodel import select
 
 from transcription.config import Settings
 from transcription.db.models import Document
+from transcription.db.models import ExecutionAttempt
 from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.providers.base import ProviderUsage
@@ -20,6 +23,15 @@ from transcription.services.store import create_job_for_document
 from transcription.services.workflows import advance_job
 
 
+async def _attempts_for_job(session, job) -> list[ExecutionAttempt]:
+    """Load execution attempts for a job; V4.7 moved evidence off JobSource."""
+    job_source_ids = [job_source.id for job_source in job.job_sources]
+    result = await session.exec(
+        select(ExecutionAttempt).where(col(ExecutionAttempt.job_source_id).in_(job_source_ids))
+    )
+    return list(result.all())
+
+
 def _jpeg_bytes(color: str = "white") -> bytes:
     output = io.BytesIO()
     Image.new("RGB", (2, 2), color=color).save(output, format="JPEG")
@@ -116,21 +128,24 @@ class TestPipelineSuccessFlow:
         assert processed is True
         assert job is not None
         assert job.status == JobStatus.TRANSCRIBED
-        assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources)
+        attempts = await _attempts_for_job(async_session, job)
+        assert any(attempt.raw_transcription == "Pipeline transcript" for attempt in attempts)
         assert job.prompt_name == "transcribe_document.md"
         assert job.user_prompt is not None
         assert job.temperature == 0.2
         assert job.top_p == 0.85
         assert any(
-            job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}}
+            attempt.normalized_metadata is not None
-            for job_source in job.job_sources
+            and attempt.normalized_metadata["finish_reason"] == "stop"
+            and attempt.normalized_metadata["usage"] == {"total_tokens": 42}
+            for attempt in attempts
         )
         assert any(
-            job_source.raw_api_response
+            attempt.sdk_response_snapshot
             == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]}
-            for job_source in job.job_sources
+            for attempt in attempts
         )
-        assert all(job_source.error_detail is None for job_source in job.job_sources)
+        assert all(attempt.error_detail is None for attempt in attempts)
 
     @pytest.mark.asyncio
     async def test_worker_transcribes_all_sources_for_multi_page_job(
@@ -191,7 +206,8 @@ class TestPipelineSuccessFlow:
         assert job.status == JobStatus.TRANSCRIBED
         assert len(job.job_sources) == 3
         assert all(job_source.status == JobSourceStatus.TRANSCRIBED for job_source in job.job_sources)
-        assert all(job_source.raw_transcription for job_source in job.job_sources)
+        attempts = await _attempts_for_job(async_session, job)
+        assert all(attempt.raw_transcription for attempt in attempts)
         assert all(
             job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources
         )
@@ -261,7 +277,8 @@ class TestPipelineSuccessFlow:
         assert len(job.job_sources) == 2
         statuses = {job_source.status for job_source in job.job_sources}
         assert statuses == {JobSourceStatus.TRANSCRIBED, JobSourceStatus.FAILED}
-        assert any(job_source.error_detail is not None for job_source in job.job_sources)
+        attempts = await _attempts_for_job(async_session, job)
+        assert any(attempt.error_detail is not None for attempt in attempts)
 
     @pytest.mark.asyncio
     async def test_worker_skips_already_transcribed_sources_on_resubmit(
@@ -293,9 +310,7 @@ class TestPipelineSuccessFlow:
         page_two = next(js for js in job.job_sources if js.source is not None and js.source.page_number == 2)
 
         page_one.status = JobSourceStatus.TRANSCRIBED
-        page_one.raw_transcription = "existing transcript"
         page_two.status = JobSourceStatus.PENDING
-        page_two.raw_transcription = None
         await services.sources.update_job_source(job_source=page_one, session=async_session)
         await services.sources.update_job_source(job_source=page_two, session=async_session)
         await services.jobs.update_job_state(job_id=job.id, status=JobStatus.QUEUED, session=async_session)
@@ -387,11 +402,10 @@ class TestPipelineFailureFlow:
         assert processed is True
         assert job is not None
         assert job.status == JobStatus.FAILED
-        assert all(job_source.raw_transcription is None for job_source in job.job_sources)
+        attempts = await _attempts_for_job(async_session, job)
-        assert any(job_source.error_detail is not None for job_source in job.job_sources)
+        assert all(attempt.raw_transcription is None for attempt in attempts)
-        error_detail = next(
+        assert any(attempt.error_detail is not None for attempt in attempts)
-            job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None
+        error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None)
-        )
         assert "pipeline provider failure" in error_detail
         assert "[internal_unexpected_error]" in error_detail
         assert "error_id=" in error_detail

tests/services/test_job_service.py

@@ -342,7 +342,6 @@ class TestJobService:
                     job_id=job.id,
                     source_id=source_one.id,
                     status=JobSourceStatus.TRANSCRIBED,
-                    raw_transcription="done",
                 )
             )
             session.add(
@@ -360,9 +359,8 @@ class TestJobService:
         refreshed = await job_service.read_job(job_id=job.id)
         statuses = {item.status for item in refreshed.job_sources}
         assert JobSourceStatus.TRANSCRIBED in statuses
-        assert JobSourceStatus.FAILED in statuses
+        assert JobSourceStatus.CANCELLED in statuses
-        pending_entry = next(item for item in refreshed.job_sources if item.status == JobSourceStatus.FAILED)
+        assert JobSourceStatus.FAILED not in statuses
-        assert pending_entry.error_detail == "Cancelled by user"
 
     @pytest.mark.asyncio
     async def test_resubmit_failed_sources_resets_only_failed(
@@ -406,8 +404,6 @@ class TestJobService:
                     job_id=job.id,
                     source_id=source_one.id,
                     status=JobSourceStatus.FAILED,
-                    raw_transcription=None,
-                    error_detail="prior error",
                 )
             )
             session.add(
@@ -415,7 +411,6 @@ class TestJobService:
                     job_id=job.id,
                     source_id=source_two.id,
                     status=JobSourceStatus.TRANSCRIBED,
-                    raw_transcription="done text",
                 )
             )
             await session.commit()
@@ -433,7 +428,6 @@ class TestJobService:
             item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 2
         )
         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 == "existing text"
         assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED
@@ -486,7 +480,6 @@ class TestJobService:
                     job_id=job.id,
                     source_id=source_two.id,
                     status=JobSourceStatus.TRANSCRIBED,
-                    raw_transcription="done text",
                 )
             )
             await session.commit()
@@ -494,6 +487,46 @@ class TestJobService:
         with pytest.raises(JobResubmitBlockedError):
             await job_service.resubmit_failed_sources(job_id=job.id)
 
+    @pytest.mark.asyncio
+    async def test_resubmit_failed_sources_includes_cancelled(
+        self,
+        job_service: JobService,
+        document_service: DocumentService,
+    ):
+        """Cancel is recoverable: cancelled pages are re-attempted on resubmit."""
+        document = Document(id=uuid4(), name="resubmit-cancelled-doc")
+        await document_service.create_document(document=document)
+
+        job = Job(document_id=document.id, status=JobStatus.FAILED)
+        await job_service.create_job(job=job)
+
+        async with job_service._session_scope() as session:
+            source = Source(
+                document_id=document.id,
+                page_number=1,
+                upload_name="resubmit-cancelled.jpg",
+                filename="stored-resubmit-cancelled.jpg",
+                file_path="/uploads/stored-resubmit-cancelled.jpg",
+                file_hash="3" * 64,
+                file_size_bytes=1,
+            )
+            session.add(source)
+            await session.flush()
+            session.add(
+                JobSource(
+                    job_id=job.id,
+                    source_id=source.id,
+                    status=JobSourceStatus.CANCELLED,
+                )
+            )
+            await session.commit()
+
+        assert await job_service.resubmit_failed_sources(job_id=job.id) == 1
+
+        refreshed = await job_service.read_job(job_id=job.id)
+        assert refreshed.status == JobStatus.QUEUED
+        assert refreshed.job_sources[0].status == JobSourceStatus.PENDING
+
     @pytest.mark.asyncio
     async def test_resubmit_failed_sources_blocks_when_processing(
         self,

tests/services/test_normalization.py

@@ -1,31 +1,19 @@
-"""Tests for V4.5 metadata-directed orientation normalization."""
+"""Tests for ingest-time orientation normalization."""
 
 import hashlib
 import io
-from pathlib import Path
 
 import pytest
 from PIL import Image
+from PIL import JpegImagePlugin
 
 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 Source
-from transcription.providers import RequestManifest
-from transcription.providers import TranscriptionResult
-from transcription.providers.evidence import SourceEvidenceReference
-from transcription.providers.evidence import build_software_context
-from transcription.services import ServiceBundle
-from transcription.services.documents import DocumentService
-from transcription.services.jobs import JobService
 from transcription.services.normalization import normalize_orientation
 from transcription.services.normalization import normalize_orientation_async
-from transcription.services.sources import SourceService
+from transcription.services.store import store_source_file
-from transcription.services.workflows import process_queued_job
 
 
-def _write_oriented_jpeg(path: Path, *, orientation: int) -> bytes:
+def _oriented_jpeg(orientation: int) -> bytes:
     image = Image.new("RGB", (2, 3))
     image.putdata(
         [
@@ -39,214 +27,133 @@ def _write_oriented_jpeg(path: Path, *, orientation: int) -> bytes:
     )
     exif = Image.Exif()
     exif[274] = orientation
-    image.save(path, format="JPEG", quality=100, subsampling=0, exif=exif)
+    buffer = io.BytesIO()
-    return path.read_bytes()
+    image.save(buffer, format="JPEG", quality=100, subsampling=0, exif=exif)
+    return buffer.getvalue()
 
 
-def _write_oriented_image(path: Path, *, orientation: int, image_format: str) -> None:
+def _oriented_image(orientation: int, image_format: str) -> bytes:
     image = Image.new("RGB", (2, 3), color="white")
     exif = Image.Exif()
     exif[274] = orientation
-    image.save(path, format=image_format, exif=exif)
+    buffer = io.BytesIO()
+    image.save(buffer, format=image_format, exif=exif)
+    return buffer.getvalue()
 
 
 @pytest.mark.unit
-def test_orientation_three_is_physically_rotated_and_metadata_removed(tmp_path):
+def test_orientation_three_is_physically_rotated_and_metadata_removed():
-    path = tmp_path / "upside-down.jpg"
+    result = normalize_orientation(_oriented_jpeg(3), media_type="image/jpeg")
-    original = _write_oriented_jpeg(path, orientation=3)
-
-    result = normalize_orientation(path, media_type="image/jpeg")
 
     assert result is not None
+    assert result.original_orientation == 3
     assert result.applied_rotation_degrees == 180
-    assert path.read_bytes() == original
+    with Image.open(io.BytesIO(result.content)) as normalized:
-    with Image.open(path) as source_image, Image.open(io.BytesIO(result.content)) as derivative:
+        assert normalized.getexif().get(274, 1) == 1
-        assert source_image.getexif()[274] == 3
+        pixel = normalized.getpixel((0, 0))
-        assert derivative.getexif().get(274, 1) == 1
-        pixel = derivative.getpixel((0, 0))
         assert isinstance(pixel, tuple)
         assert pixel[2] > pixel[0]
 
 
 @pytest.mark.unit
-def test_orientation_one_is_noop(tmp_path):
+def test_orientation_one_is_noop():
-    path = tmp_path / "upright.jpg"
+    assert normalize_orientation(_oriented_jpeg(1), media_type="image/jpeg") is None
-    _write_oriented_jpeg(path, orientation=1)
-
-    assert normalize_orientation(path, media_type="image/jpeg") is None
 
 
+@pytest.mark.unit
+def test_normalization_is_idempotent():
+    once = normalize_orientation(_oriented_jpeg(3), media_type="image/jpeg")
+
+    assert once is not None
+    assert normalize_orientation(once.content, media_type="image/jpeg") is None
+
+
+@pytest.mark.unit
+def test_jpeg_reencode_reuses_source_quantization_tables():
+    """Reusing the source tables is what keeps the rewrite small and near-lossless."""
+    original = _oriented_jpeg(3)
+    result = normalize_orientation(original, media_type="image/jpeg")
+
+    assert result is not None
+    with (
+        Image.open(io.BytesIO(original)) as before,
+        Image.open(io.BytesIO(result.content)) as after,
+    ):
+        assert isinstance(before, JpegImagePlugin.JpegImageFile)
+        assert isinstance(after, JpegImagePlugin.JpegImageFile)
+        assert after.quantization == before.quantization
+
+
+@pytest.mark.unit
 @pytest.mark.parametrize(
-    ("filename", "image_format", "media_type"),
+    ("image_format", "media_type"),
-    [
+    [("PNG", "image/png"), ("TIFF", "image/tiff")],
-        ("oriented.png", "PNG", "image/png"),
-        ("oriented.tiff", "TIFF", "image/tiff"),
-    ],
 )
-def test_supported_non_jpeg_orientation_is_normalized(
+def test_supported_non_jpeg_orientation_is_normalized(image_format, media_type):
-    tmp_path,
+    result = normalize_orientation(_oriented_image(6, image_format), media_type=media_type)
-    filename,
-    image_format,
-    media_type,
-):
-    path = tmp_path / filename
-    _write_oriented_image(path, orientation=6, image_format=image_format)
-
-    result = normalize_orientation(path, media_type=media_type)
 
     assert result is not None
     assert result.applied_rotation_degrees == 90
-    assert (result.derivative_width, result.derivative_height) == (3, 2)
+    with Image.open(io.BytesIO(result.content)) as normalized:
-    with Image.open(io.BytesIO(result.content)) as derivative:
+        assert normalized.getexif().get(274, 1) == 1
-        assert derivative.getexif().get(274, 1) == 1
 
 
-@pytest.mark.integration
+@pytest.mark.unit
-@pytest.mark.asyncio
+def test_unsupported_media_type_is_left_alone():
-async def test_resolve_provider_input_persists_exact_derivative(default_session_factory, tmp_path):
+    assert normalize_orientation(b"not-an-image", media_type="application/pdf") is None
-    source_path = tmp_path / "source.jpg"
-    original = _write_oriented_jpeg(source_path, orientation=3)
-    settings = Settings(
-        openrouter_api_key="test-key",
-        artifact_dir=tmp_path / "artifacts",
-        provider_models=None,
-    )
-    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="Oriented"))
-    source = await sources.create_source(
-        Source(
-            document_id=document.id,
-            page_number=1,
-            upload_name="source.jpg",
-            filename="source.jpg",
-            file_path=str(source_path),
-            file_hash=hashlib.sha256(original).hexdigest(),
-            file_size_bytes=len(original),
-        )
-    )
-
-    provider_input = await sources.resolve_provider_input(source)
-    artifacts = await sources.list_processing_artifacts(source_id=source.id)
-
-    assert source_path.read_bytes() == original
-    assert provider_input.derivative_id == artifacts[0].id
-    assert provider_input.path.read_bytes() != original
-    assert hashlib.sha256(provider_input.path.read_bytes()).hexdigest() == provider_input.digest_sha256
-    coordinate_metadata = artifacts[0].coordinate_metadata
-    assert coordinate_metadata is not None
-    assert coordinate_metadata["original_orientation"] == 3
-
-
-@pytest.mark.integration
-@pytest.mark.asyncio
-async def test_worker_sends_exact_derivative_and_links_attempt_evidence(
-    default_session_factory,
-    tmp_path,
-    monkeypatch,
-):
-    source_path = tmp_path / "source.jpg"
-    original = _write_oriented_jpeg(source_path, orientation=3)
-    prompt_dir = tmp_path / "prompts"
-    prompt_dir.mkdir()
-    (prompt_dir / "transcribe_document.md").write_text("Transcribe verbatim.", encoding="utf-8")
-    settings = Settings(
-        openrouter_api_key="test-key",
-        artifact_dir=tmp_path / "artifacts",
-        prompt_dir=prompt_dir,
-        provider_models=None,
-    )
-    services = ServiceBundle(
-        documents=DocumentService(session_factory=default_session_factory, settings=settings),
-        jobs=JobService(session_factory=default_session_factory, settings=settings),
-        sources=SourceService(session_factory=default_session_factory, settings=settings),
-    )
-    document = await services.documents.create_document(Document(name="Pipeline"))
-    source = await services.sources.create_source(
-        Source(
-            document_id=document.id,
-            page_number=1,
-            upload_name="source.jpg",
-            filename="source.jpg",
-            file_path=str(source_path),
-            file_hash=hashlib.sha256(original).hexdigest(),
-            file_size_bytes=len(original),
-        )
-    )
-    job = await services.jobs.create_job(Job(document_id=document.id))
-    await services.sources.create_job_source(JobSource(job_id=job.id, source_id=source.id))
-    loaded = await services.jobs.read_job(job.id)
-    captured: dict[str, object] = {}
-
-    async def fake_transcribe(
-        image_path,
-        *,
-        prompt_name,
-        prompt_text,
-        temperature,
-        top_p,
-        settings,
-        provider,
-        source_reference,
-        requested_model,
-    ):
-        _ = (prompt_name, temperature, top_p, settings, provider, requested_model)
-        image_bytes = Path(image_path).read_bytes()
-        captured["bytes"] = image_bytes
-        captured["source_reference"] = source_reference
-        manifest = RequestManifest(
-            provider="fixture",
-            requested_model="fixture/model",
-            request={"model": "fixture/model"},
-            source=source_reference,
-            optional_parameter_states={"temperature": "omitted", "top_p": "omitted"},
-            prompt_content=prompt_text,
-            prompt_sha256=hashlib.sha256(prompt_text.encode()).hexdigest(),
-            timeout_seconds=20,
-            retry_policy="none",
-            software=build_software_context(
-                adapter_name="fixture",
-                adapter_version="1",
-                client_library="transcription",
-            ),
-        )
-        return TranscriptionResult(
-            text="[document body typewritten]\nDamaged \ufffd text",
-            provider="fixture",
-            model="fixture/model",
-            request_manifest=manifest,
-        )
-
-    monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", fake_transcribe)
-
-    await process_queued_job(job=loaded, services=services, settings=settings)
-
-    attempts = await services.sources.list_execution_attempts(source_id=source.id)
-    artifacts = await services.sources.list_processing_artifacts(source_id=source.id)
-    source_reference = captured["source_reference"]
-    assert isinstance(source_reference, SourceEvidenceReference)
-    captured_bytes = captured["bytes"]
-    assert isinstance(captured_bytes, bytes)
-    assert source_path.read_bytes() == original
-    assert hashlib.sha256(captured_bytes).hexdigest() == source_reference.digest_sha256
-    assert source_reference.derivative_id is not None
-    assert {artifact.artifact_type for artifact in artifacts} == {
-        "orientation_normalized_model_input",
-        "transcription_quality_warnings",
-    }
-    assert {artifact.execution_attempt_id for artifact in artifacts} == {attempts[0].id}
 
 
 @pytest.mark.asyncio
-async def test_async_wrapper_matches_sync_result_and_precomputes_digest(tmp_path):
+async def test_async_wrapper_matches_sync_result():
-    """[MED-01]: Pillow work runs off the event loop and hashes its own output."""
+    """[MED-01]: Pillow work runs off the event loop."""
-    path = tmp_path / "async-upside-down.jpg"
+    original = _oriented_jpeg(3)
-    _write_oriented_jpeg(path, orientation=3)
 
-    result = await normalize_orientation_async(path, media_type="image/jpeg")
+    result = await normalize_orientation_async(original, media_type="image/jpeg")
-    expected = normalize_orientation(path, media_type="image/jpeg")
+    expected = normalize_orientation(original, media_type="image/jpeg")
 
     assert result is not None
     assert expected is not None
     assert result.content == expected.content
-    assert result.digest_sha256 == hashlib.sha256(result.content).hexdigest()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.parametrize("orientation", [3, 6, 8])
+async def test_stored_source_never_retains_exif_orientation(tmp_path, orientation):
+    settings = Settings(
+        openrouter_api_key="test-key",
+        upload_dir=tmp_path / "uploads",
+        provider_models=None,
+    )
+
+    stored = await store_source_file(
+        filename="page.jpg",
+        file_bytes=_oriented_jpeg(orientation),
+        settings=settings,
+    )
+
+    stored_bytes = stored.path.read_bytes()
+    with Image.open(io.BytesIO(stored_bytes)) as image:
+        assert image.getexif().get(274, 1) == 1
+    assert stored.file_hash == hashlib.sha256(stored_bytes).hexdigest()
+    assert stored.file_size_bytes == len(stored_bytes)
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_upright_source_is_stored_byte_for_byte(tmp_path):
+    settings = Settings(
+        openrouter_api_key="test-key",
+        upload_dir=tmp_path / "uploads",
+        provider_models=None,
+    )
+    original = _oriented_jpeg(1)
+
+    stored = await store_source_file(
+        filename="page.jpg",
+        file_bytes=original,
+        settings=settings,
+    )
+
+    assert stored.path.read_bytes() == original
+    assert stored.file_hash == hashlib.sha256(original).hexdigest()

tests/services/test_v2_crud.py

@@ -322,6 +322,10 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
 
     stored_rows = await transcriptions.list_job_sources(job_id=job.id)
     assert len(stored_rows) == 1
-    assert stored_rows[0].raw_transcription == "provider transcript"
+    assert stored_rows[0].status == JobSourceStatus.TRANSCRIBED
-    assert stored_rows[0].ai_metadata == metadata
+
-    assert stored_rows[0].raw_api_response == raw_payload
+    attempt = await transcriptions.read_latest_execution_attempt(job_source_id=stored_rows[0].id)
+    assert attempt is not None
+    assert attempt.attempt.raw_transcription == "provider transcript"
+    assert attempt.attempt.normalized_metadata == metadata
+    assert attempt.attempt.sdk_response_snapshot == raw_payload

tests/services/test_workflows_reliability.py

@@ -5,9 +5,12 @@ from pathlib import Path
 from uuid import uuid4
 
 import pytest
+from sqlmodel import col
+from sqlmodel import select
 
 from transcription.config import Settings
 from transcription.db.models import Document
+from transcription.db.models import ExecutionAttempt
 from transcription.db.models import Job
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
@@ -87,9 +90,18 @@ class TestWorkflowReliability:
 
         assert result is not None
         assert result.status == JobStatus.FAILED
-        assert result.error_detail is not None
+
-        assert "timed out" in result.error_detail.lower()
+        async with services.jobs._session_scope() as session:
-        assert "20.0s" in result.error_detail
+            attempts = (
+                await session.execute(
+                    select(ExecutionAttempt).where(
+                        col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
+                    )
+                )
+            ).scalars().all()
+        error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None)
+        assert "timed out" in error_detail.lower()
+        assert "20.0s" in error_detail
 
     @pytest.mark.asyncio
     async def test_completed_page_is_committed_before_next_provider_call_finishes(

tests/test_config.py

@@ -12,10 +12,10 @@ from transcription.config import parse_cli_settings
 
 
 def _make_settings(**overrides: Any) -> Settings:
-    """Build a Settings instance with a dummy API key unless overridden."""
+    """Build a Settings instance with a dummy API key, isolated from any local .env."""
     defaults: dict[str, Any] = {"openrouter_api_key": "test-key-abc123", "provider_models": None}
     defaults.update(overrides)
-    return Settings(**defaults)
+    return Settings(_env_file=None, **defaults)
 
 
 class TestSettingsLoading:

tests/test_db.py

@@ -45,7 +45,6 @@ async def test_create_all_creates_expected_tables(tmp_path):
         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()

tests/test_models.py

@@ -211,24 +211,17 @@ class TestPersonAndDocumentPersonModel:
 
 
 class TestJobSourceModel:
-    def test_job_source_persists_json_payloads(self, session):
+    def test_job_source_is_a_queue_row_not_an_evidence_row(self, session):
+        """V4.7: job_source carries only queue state; evidence lives on execution_attempt."""
         document = _persist_document(session)
         job = _persist_job(session, document)
         source = _persist_source(session, document)
-        job_source = _persist_job_source(
+        job_source = _persist_job_source(session, job, source)
-            session,
-            job,
-            source,
-            raw_transcription="Page transcript",
-            ai_metadata={"confidence": 0.91, "boxes": [{"x": 1, "y": 2}]},
-            raw_api_response={"provider": "test"},
-        )
 
         fetched = session.get(JobSource, job_source.id)
         assert fetched is not None
         assert fetched.status == JobSourceStatus.PENDING
-        assert fetched.ai_metadata == {"confidence": 0.91, "boxes": [{"x": 1, "y": 2}]}
+        assert set(JobSource.model_fields) == {"id", "job_id", "source_id", "status"}
-        assert fetched.raw_api_response == {"provider": "test"}
 
 
 class TestRelationships:

tests/test_v42_evidence.py

@@ -19,7 +19,6 @@ 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
@@ -28,9 +27,7 @@ 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
 
 
@@ -249,34 +246,9 @@ async def test_attempts_are_append_only_and_exported_with_integrity(default_sess
     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 _json_object(export["source"])["digest_sha256"] == "a" * 64
     assert [_json_object(item)["attempt_number"] for item in _json_array(export["attempts"])] == [1, 2]
-    assert _json_object(_json_array(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)
@@ -285,7 +257,6 @@ async def test_attempts_are_append_only_and_exported_with_integrity(default_sess
     latest_job_source = detail.latest_job_source
     assert latest_job_source is not None
     assert latest_job_source.execution_attempts == []
-    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.attempt_number == 2
@@ -304,89 +275,6 @@ def test_benchmark_scoring_preserves_literal_differences():
     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"

tests/ui/conftest.py

@@ -23,6 +23,7 @@ from transcription.db import session as db_session_module
 from transcription.db import session_scope
 from transcription.db.models import Document
 from transcription.db.models import DocumentPerson
+from transcription.db.models import ExecutionAttempt
 from transcription.db.models import Job
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
@@ -131,17 +132,30 @@ async def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., Awai
             await session.flush()
 
             if transcription_text is not None or error_detail is not None:
+                outcome = JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED
+                job_source = JobSource(job_id=job.id, source_id=source.id, status=outcome)
+                session.add(job_source)
+                await session.flush()
+
+                # V4.7: evidence lives on execution_attempt, not job_source.
+                executed_at = datetime.now(UTC)
                 session.add(
-                    JobSource(
+                    ExecutionAttempt(
+                        job_source_id=job_source.id,
                         job_id=job.id,
                         source_id=source.id,
-                        status=(
+                        attempt_number=1,
-                            JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED
+                        status=outcome,
-                        ),
+                        provider="openrouter",
+                        model="google/gemini-2.5-flash",
+                        response_received=transcription_text is not None,
+                        sdk_response_snapshot=raw_api_response,
+                        normalized_metadata=ai_metadata,
                         raw_transcription=transcription_text,
                         error_detail=error_detail,
-                        ai_metadata=ai_metadata,
+                        started_at=executed_at,
-                        raw_api_response=raw_api_response,
+                        finished_at=executed_at,
+                        duration_ms=0,
                     )
                 )
 

tests/ui/test_jobs_page.py

@@ -125,7 +125,7 @@ class TestJobsPageRendering:
 
         assert response.status_code == 200
         assert "Resubmit Job" in response.text
-        assert "Failed Sources:" in response.text
+        assert "Resubmittable Sources:" in response.text
         assert "Resubmit now" in response.text
 
     @pytest.mark.asyncio

tests/ui/test_sources_page.py

@@ -7,9 +7,11 @@ import pytest
 from sqlmodel import select
 
 from transcription.db import session_scope
+from transcription.db.loading import orm_attribute
 from transcription.db.loading import selectinload
 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 JobStatus
 from transcription.db.models import Source
@@ -56,7 +58,7 @@ class TestSourceModelProperties:
                     select(Source)
                     .options(
                         selectinload(Source.document),
-                        selectinload(Source.job_sources),
+                        selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)),
                     )
                     .where(Source.document_id == job.document_id)
                 )
@@ -218,8 +220,8 @@ class TestSourcesPageRendering:
         assert "Save revision" in response.text
         assert "Previous Page" in response.text
         assert "Next Page" in response.text
-        assert "AI Metadata" in response.text
+        assert "Normalized Metadata" in response.text
-        assert "Raw API Response" in response.text
+        assert "OpenRouter SDK Response Snapshot" in response.text
         assert "finish_reason" in response.text
         assert "response-123" in response.text
 
@@ -261,7 +263,6 @@ class TestSourcesPageRendering:
         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):

tools/migrate_v45_to_v46.py

@@ -1,329 +0,0 @@
-"""One-time migration of a V4.5 database into the re-leveled V4.6 schema.
-
-V4.6 re-levels the schema from the current SQLModel metadata rather than
-running a chain of hand-rolled upgrade functions. The column sets are
-unchanged; what changed is index coverage ([HIGH-04]), the ``use_alter``
-break in the ``source``/``execution_attempt`` foreign key cycle, and the
-relationship loading strategy ([CRIT-02]). This script therefore performs a
-faithful, foreign-key-ordered row copy.
-
-Design notes:
-
-- The backup is read with plain ``sqlite3`` rather than through the ORM. The
-  V4.5 file is not guaranteed to satisfy the V4.6 mappers, and reading raw
-  rows means no relationship is ever traversed, so ``lazy="raise"`` cannot
-  bite.
-- The target is written through SQLAlchemy Core against the live metadata, so
-  the same script works against PostgreSQL when that cutover happens.
-- Identity is preserved exactly: UUIDs, digests, timestamps, attempt numbers,
-  and ``preferred_execution_attempt_id`` selections carry across unchanged.
-  No evidence payload is reinterpreted, normalized, or regenerated.
-- No on-disk Source file, portrait, or artifact file is read for writing or
-  modified. ``--verify-artifacts`` reads artifact files, but only to hash
-  them.
-- The script is idempotent: a row whose primary key already exists in the
-  target is skipped, never updated. It is never invoked from application
-  startup and never runs in the test suite.
-
-Usage::
-
-    python tools/migrate_v45_to_v46.py --dry-run
-    python tools/migrate_v45_to_v46.py --verify-artifacts
-"""
-
-from __future__ import annotations
-
-import argparse
-import json
-import sqlite3
-import sys
-from collections.abc import Iterator
-from collections.abc import Sequence
-from datetime import date
-from datetime import datetime
-from pathlib import Path
-from typing import Any
-from uuid import UUID
-
-from sqlalchemy import Column
-from sqlalchemy import Table
-from sqlalchemy import create_engine
-from sqlalchemy import insert
-from sqlalchemy import inspect as sqlalchemy_inspect
-from sqlalchemy import select
-from sqlalchemy import update
-from sqlalchemy.engine import Connection
-from sqlmodel import SQLModel
-
-from transcription.config import Settings
-from transcription.config import get_settings
-from transcription.db import models as _models  # noqa: F401  (registers every table)
-from transcription.db.engine import get_database_url
-
-DEFAULT_BACKUP = Path("data/transcription.db.pre-v46.bak")
-
-#: ``source.preferred_execution_attempt_id`` points at ``execution_attempt``,
-#: which points back at ``source``. The cycle is broken with ``use_alter`` in
-#: the metadata, so ``source`` rows are inserted with the column cleared and
-#: the selections are replayed once ``execution_attempt`` is populated.
-DEFERRED_TABLE = "source"
-DEFERRED_COLUMN = "preferred_execution_attempt_id"
-
-#: Row counts the V4.5 backup is expected to carry, used as a pre-flight guard
-#: so the script cannot silently run against the wrong file.
-EXPECTED_SOURCE_COUNTS = {
-    "document": 8,
-    "document_person": 11,
-    "document_type": 7,
-    "execution_attempt": 80,
-    "job": 11,
-    "job_source": 79,
-    "person": 5,
-    "person_role": 3,
-    "processing_artifact": 2,
-    "source": 76,
-}
-
-
-def _coerce(column: Column[Any], value: object) -> object:
-    """Convert a raw SQLite value into what the target column's type binds.
-
-    SQLite hands back strings and integers; the V4.6 columns bind ``UUID``,
-    ``datetime``, ``date``, ``bool``, enum members, and decoded JSON. The
-    conversion is lossless in both directions.
-    """
-    if value is None:
-        return None
-
-    match type(column.type).__name__:
-        case "Uuid":
-            return value if isinstance(value, UUID) else UUID(str(value))
-        case "DateTime":
-            return value if isinstance(value, datetime) else datetime.fromisoformat(str(value))
-        case "Date":
-            return value if isinstance(value, date) else date.fromisoformat(str(value))
-        case "Boolean":
-            return bool(value)
-        case "JSONBCompat":
-            if isinstance(value, str | bytes | bytearray):
-                return json.loads(value)
-            return value
-        case "Enum":
-            enum_class = getattr(column.type, "enum_class", None)
-            if enum_class is None:
-                return value
-            # The same JobSourceStatus enum is persisted by value on
-            # job_source.status and by name on execution_attempt.status,
-            # because only the former declares values_callable. Accept either
-            # spelling so the copy round-trips both columns faithfully.
-            try:
-                return enum_class(value)
-            except ValueError:
-                return enum_class[str(value)]
-        case _:
-            return value
-
-
-def _read_table(backup: sqlite3.Connection, table: Table) -> list[dict[str, object]]:
-    """Read every row of ``table`` from the backup, coerced for the target."""
-    names = [column.name for column in table.columns]
-    quoted = ", ".join(f'"{name}"' for name in names)
-    rows: list[dict[str, object]] = []
-    for raw in backup.execute(f'select {quoted} from "{table.name}"'):
-        rows.append({name: _coerce(table.columns[name], raw[index]) for index, name in enumerate(names)})
-    return rows
-
-
-def _primary_key(table: Table) -> Column[Any]:
-    columns = list(table.primary_key.columns)
-    if len(columns) != 1:
-        message = f"{table.name} does not have a single-column primary key"
-        raise RuntimeError(message)
-    return columns[0]
-
-
-def _existing_keys(connection: Connection, table: Table) -> set[object]:
-    key = _primary_key(table)
-    return set(connection.execute(select(key)).scalars().all())
-
-
-def _chunked(rows: Sequence[dict[str, object]], size: int = 200) -> Iterator[Sequence[dict[str, object]]]:
-    for start in range(0, len(rows), size):
-        yield rows[start : start + size]
-
-
-def _verify_artifacts(settings: Settings) -> int:
-    """Re-hash every migrated artifact through the service's own verifier."""
-    from transcription.db.models import ProcessingArtifact
-    from transcription.services.sources import SourceService
-
-    engine = create_engine(_sync_url(settings))
-    with engine.connect() as connection:
-        rows = connection.execute(select(SQLModel.metadata.tables["processing_artifact"])).mappings().all()
-    engine.dispose()
-
-    service = SourceService(settings=settings)
-    artifacts = [ProcessingArtifact(**dict(row)) for row in rows]
-    # Reuses the application's own integrity check so the migration cannot
-    # disagree with what the running app considers a valid artifact.
-    service._verify_artifacts_integrity(artifacts)
-    return len(artifacts)
-
-
-def _sync_url(settings: Settings) -> str:
-    """Return the target database URL with any async driver stripped."""
-    url = get_database_url(settings)
-    return url.replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "")
-
-
-def _preflight(backup: sqlite3.Connection, *, strict: bool) -> None:
-    actual = {
-        name: backup.execute(f'select count(*) from "{name}"').fetchone()[0]
-        for name in EXPECTED_SOURCE_COUNTS
-    }
-    mismatched = {
-        name: (count, EXPECTED_SOURCE_COUNTS[name])
-        for name, count in actual.items()
-        if count != EXPECTED_SOURCE_COUNTS[name]
-    }
-    if not mismatched:
-        return
-    detail = ", ".join(f"{name}: found {found}, expected {want}" for name, (found, want) in sorted(mismatched.items()))
-    message = f"Backup row counts do not match the recorded V4.5 snapshot ({detail})"
-    if strict:
-        raise RuntimeError(message)
-    print(f"WARNING: {message}", file=sys.stderr)
-
-
-def _copy_tables(
-    connection: Connection,
-    payload: dict[str, list[dict[str, object]]],
-    *,
-    dry_run: bool,
-) -> tuple[int, dict[object, object]]:
-    """Insert every missing row, deferring the cyclic foreign key column."""
-    deferred: dict[object, object] = {}
-    inserted_total = 0
-
-    for table in SQLModel.metadata.sorted_tables:
-        rows = payload[table.name]
-        existing = set() if dry_run else _existing_keys(connection, table)
-        key_name = _primary_key(table).name
-
-        pending = [row for row in rows if row[key_name] not in existing]
-
-        if table.name == DEFERRED_TABLE:
-            for row in pending:
-                selection = row[DEFERRED_COLUMN]
-                if selection is not None:
-                    deferred[row[key_name]] = selection
-                    row[DEFERRED_COLUMN] = None
-
-        if pending and not dry_run:
-            for chunk in _chunked(pending):
-                connection.execute(insert(table), list(chunk))
-
-        inserted_total += len(pending)
-        print(f"  {table.name:24} insert={len(pending):<5} skip={len(rows) - len(pending)}")
-
-    return inserted_total, deferred
-
-
-def _replay_deferred(connection: Connection, deferred: dict[object, object], *, dry_run: bool) -> None:
-    """Restore the preferred-attempt selections held back by the FK cycle."""
-    if not deferred:
-        return
-    print(f"  replaying {len(deferred)} deferred {DEFERRED_TABLE}.{DEFERRED_COLUMN} selection(s)")
-    if dry_run:
-        return
-    source = SQLModel.metadata.tables[DEFERRED_TABLE]
-    key = _primary_key(source)
-    for source_id, attempt_id in deferred.items():
-        connection.execute(update(source).where(key == source_id).values({DEFERRED_COLUMN: attempt_id}))
-
-
-def _report_counts(connection: Connection) -> None:
-    print("\nPost-migration row counts:")
-    for table in SQLModel.metadata.sorted_tables:
-        actual = len(connection.execute(select(_primary_key(table))).all())
-        expected = EXPECTED_SOURCE_COUNTS.get(table.name)
-        flag = "" if expected is None or actual == expected else f"  <-- expected {expected}"
-        print(f"  {table.name:24} {actual}{flag}")
-
-
-def _load_payload(backup_path: Path, *, strict_counts: bool) -> dict[str, list[dict[str, object]]]:
-    if not backup_path.is_file():
-        message = f"Backup database not found: {backup_path}"
-        raise FileNotFoundError(message)
-    backup = sqlite3.connect(f"file:{backup_path}?mode=ro", uri=True)
-    try:
-        _preflight(backup, strict=strict_counts)
-        return {table.name: _read_table(backup, table) for table in SQLModel.metadata.sorted_tables}
-    finally:
-        backup.close()
-
-
-def migrate(*, backup_path: Path, settings: Settings, dry_run: bool, strict_counts: bool) -> int:
-    """Copy every row from the V4.5 backup into the re-leveled schema."""
-    payload = _load_payload(backup_path, strict_counts=strict_counts)
-
-    engine = create_engine(_sync_url(settings))
-    try:
-        if not sqlalchemy_inspect(engine).has_table("document"):
-            print("Target schema is empty; creating it from the current metadata.")
-            if not dry_run:
-                SQLModel.metadata.create_all(engine)
-
-        with engine.begin() as connection:
-            inserted_total, deferred = _copy_tables(connection, payload, dry_run=dry_run)
-            _replay_deferred(connection, deferred, dry_run=dry_run)
-
-        if dry_run:
-            print("\nDry run: no rows were written.")
-            return inserted_total
-
-        with engine.connect() as connection:
-            _report_counts(connection)
-    finally:
-        engine.dispose()
-
-    return inserted_total
-
-
-def main(argv: Sequence[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
-    parser.add_argument("--backup", type=Path, default=DEFAULT_BACKUP, help="V4.5 database to read from")
-    parser.add_argument("--dry-run", action="store_true", help="Report what would be copied without writing")
-    parser.add_argument(
-        "--allow-count-mismatch",
-        action="store_true",
-        help="Warn instead of aborting when the backup row counts differ from the recorded snapshot",
-    )
-    parser.add_argument(
-        "--verify-artifacts",
-        action="store_true",
-        help="Re-hash every migrated processing artifact after the copy",
-    )
-    args = parser.parse_args(argv)
-
-    settings = get_settings()
-    print(f"Source: {args.backup}")
-    print(f"Target: {_sync_url(settings)}\n")
-
-    inserted = migrate(
-        backup_path=args.backup,
-        settings=settings,
-        dry_run=args.dry_run,
-        strict_counts=not args.allow_count_mismatch,
-    )
-
-    if args.verify_artifacts and not args.dry_run:
-        verified = _verify_artifacts(settings)
-        print(f"\nArtifact integrity verified for {verified} artifact(s).")
-
-    print(f"\nDone. {inserted} row(s) inserted.")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

tools/migrate_v46_to_v47.py

@@ -0,0 +1,339 @@
+"""One-time migration of a V4.6 database into the V4.7 schema.
+
+V4.7 is an architectural cleanup: no new user-facing behaviour, but three
+structural changes plus a one-time image backfill. This tool carries all of
+them, and is built up phase by phase so the live database stays usable at
+every phase boundary.
+
+Steps, in execution order:
+
+1. Rotate every stored Source image that still carries a supported EXIF
+   orientation, in place, and update ``source.file_hash`` and
+   ``source.file_size_bytes`` to describe the rewritten file.
+2. Drop the ``processing_artifact`` table and delete its external files.
+3. Rewrite ``execution_attempt.status`` from enum *names* to enum *values*, so
+   it compares equal to ``job_source.status`` (defect [45]).
+4. Drop the five evidence columns from ``job_source``, leaving it a pure work
+   queue of ``id``, ``job_id``, ``source_id`` and ``status``.
+
+Design notes:
+
+- The image rewrite reuses the application's own
+  :func:`~transcription.services.normalization.normalize_orientation`, so the
+  backfilled bytes are byte-identical to what ingest would now produce. It
+  reuses the source quantization tables and subsampling rather than
+  re-quantizing, which is both smaller and higher fidelity than a fixed
+  quality setting.
+- The hash and size are rewritten alongside the file. After V4.7 the evidence
+  digest is derived straight from ``source.file_hash``, so leaving it
+  describing the pre-rotation bytes would silently invalidate every future
+  export.
+- The database is read and written through SQLAlchemy Core against the live
+  metadata, so the same script works against PostgreSQL when that cutover
+  happens. Raw DDL is used only for the table drop, which has no Core
+  equivalent that is safe to express against deleted metadata.
+- The script is idempotent, keyed on state rather than on a version marker:
+  an image with no supported orientation tag is skipped, and a table that is
+  already absent is skipped. It is never invoked from application startup and
+  never runs in the test suite.
+- **The application must not be running.** The image rewrite is not atomic
+  with the row update, and SQLite will refuse the schema change while another
+  connection holds the database.
+
+Usage::
+
+    python tools/migrate_v46_to_v47.py --dry-run
+    python tools/migrate_v46_to_v47.py
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import sys
+from collections.abc import Sequence
+from pathlib import Path
+
+from sqlalchemy import bindparam
+from sqlalchemy import create_engine
+from sqlalchemy import inspect as sqlalchemy_inspect
+from sqlalchemy import select
+from sqlalchemy import text
+from sqlalchemy import update
+from sqlalchemy.engine import Connection
+from sqlmodel import SQLModel
+
+from transcription.config import Settings
+from transcription.config import get_settings
+from transcription.db import models as _models  # noqa: F401  (registers every table)
+from transcription.db.engine import get_database_url
+from transcription.db.models import JobSourceStatus
+from transcription.services.normalization import normalize_orientation
+from transcription.services.sources import source_mime_type
+
+#: Row counts the V4.6 database is expected to carry, used as a pre-flight
+#: guard so the script cannot silently run against the wrong file.
+EXPECTED_ROW_COUNTS = {
+    "document": 8,
+    "document_person": 11,
+    "document_type": 7,
+    "execution_attempt": 80,
+    "job": 11,
+    "job_source": 79,
+    "person": 5,
+    "person_role": 3,
+    "source": 76,
+}
+
+ARTIFACT_TABLE = "processing_artifact"
+
+#: Evidence columns removed from ``job_source`` in V4.7. Every one of them is
+#: duplicated byte-for-byte by ``execution_attempt`` across all 77 rows that
+#: carry evidence, so no information is lost by dropping them.
+JOB_SOURCE_DROPPED_COLUMNS = (
+    "raw_transcription",
+    "ai_metadata",
+    "raw_api_response",
+    "error_detail",
+    "executed_at",
+)
+
+#: The V4.6 default for the deleted ``Settings.artifact_dir``. The setting no
+#: longer exists, so the historical location is recorded here instead.
+DEFAULT_ARTIFACT_DIR = Path("data/artifacts")
+
+
+def _sync_url(settings: Settings) -> str:
+    """Return the target database URL with any async driver stripped."""
+    url = get_database_url(settings)
+    return url.replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "")
+
+
+def _preflight(connection: Connection, *, strict: bool) -> None:
+    inspector = sqlalchemy_inspect(connection)
+    present = set(inspector.get_table_names())
+    mismatched: dict[str, tuple[object, int]] = {}
+    for name, expected in EXPECTED_ROW_COUNTS.items():
+        if name not in present:
+            mismatched[name] = ("missing", expected)
+            continue
+        actual = connection.execute(text(f'select count(*) from "{name}"')).scalar_one()
+        if actual != expected:
+            mismatched[name] = (actual, expected)
+    if not mismatched:
+        return
+    detail = ", ".join(f"{name}: found {found}, expected {want}" for name, (found, want) in sorted(mismatched.items()))
+    message = f"Database row counts do not match the recorded V4.6 snapshot ({detail})"
+    if strict:
+        raise RuntimeError(message)
+    print(f"WARNING: {message}", file=sys.stderr)
+
+
+def rotate_stored_images(connection: Connection, *, dry_run: bool) -> int:
+    """Step 1: rewrite every mis-oriented stored image and its recorded digest."""
+    source = SQLModel.metadata.tables["source"]
+    rows = connection.execute(
+        select(source.c.id, source.c.file_path, source.c.filename)
+    ).all()
+
+    rotated = 0
+    missing = 0
+    for source_id, file_path, filename in rows:
+        path = Path(str(file_path))
+        if not path.is_file():
+            print(f"  WARNING: source file not found, skipped: {path}", file=sys.stderr)
+            missing += 1
+            continue
+
+        content = path.read_bytes()
+        normalized = normalize_orientation(content, media_type=source_mime_type(str(filename)))
+        if normalized is None:
+            continue
+
+        rotated += 1
+        print(
+            f"  {path.name} orientation={normalized.original_orientation} "
+            f"rotation={normalized.applied_rotation_degrees} "
+            f"{len(content)} -> {len(normalized.content)} bytes"
+        )
+        if dry_run:
+            continue
+
+        path.write_bytes(normalized.content)
+        connection.execute(
+            update(source)
+            .where(source.c.id == source_id)
+            .values(
+                file_hash=hashlib.sha256(normalized.content).hexdigest(),
+                file_size_bytes=len(normalized.content),
+            )
+        )
+
+    print(f"  rotated={rotated} upright={len(rows) - rotated - missing} missing={missing}")
+    return rotated
+
+
+def drop_processing_artifacts(connection: Connection, artifact_dir: Path, *, dry_run: bool) -> int:
+    """Step 2: drop the artifact table and delete the files it referenced."""
+    inspector = sqlalchemy_inspect(connection)
+    if ARTIFACT_TABLE not in set(inspector.get_table_names()):
+        print(f"  {ARTIFACT_TABLE} already absent")
+        return 0
+
+    references = [
+        str(row[0])
+        for row in connection.execute(
+            text(f'select external_reference from "{ARTIFACT_TABLE}" where external_reference is not null')
+        )
+    ]
+    count = connection.execute(text(f'select count(*) from "{ARTIFACT_TABLE}"')).scalar_one()
+    print(f"  dropping {ARTIFACT_TABLE} ({count} row(s), {len(references)} external file(s))")
+
+    if dry_run:
+        return count
+
+    connection.execute(text(f'drop table "{ARTIFACT_TABLE}"'))
+
+    artifact_root = artifact_dir.resolve()
+    for reference in references:
+        relative = Path(reference)
+        if relative.is_absolute() or ".." in relative.parts:
+            print(f"  WARNING: skipped unsafe artifact reference: {reference}", file=sys.stderr)
+            continue
+        artifact_path = (artifact_root / relative).resolve()
+        if artifact_root not in artifact_path.parents:
+            print(f"  WARNING: skipped artifact outside root: {reference}", file=sys.stderr)
+            continue
+        artifact_path.unlink(missing_ok=True)
+        parent = artifact_path.parent
+        if parent != artifact_root and parent.is_dir() and not any(parent.iterdir()):
+            parent.rmdir()
+
+    return count
+
+
+def normalize_attempt_status(connection: Connection, *, dry_run: bool) -> int:
+    """Step 3: rewrite ``execution_attempt.status`` from enum names to values.
+
+    Defect [45]: ``execution_attempt.status`` was declared without
+    ``values_callable``, so SQLAlchemy persisted enum *names* ('TRANSCRIBED')
+    while ``job_source.status`` persisted *values* ('transcribed'). The two
+    columns never compared equal on a single one of the 79 rows. The model
+    declaration is fixed in V4.7; the stored rows are fixed here.
+    """
+    name_to_value = {member.name: member.value for member in JobSourceStatus}
+    recognised = sorted(set(name_to_value) | set(name_to_value.values()))
+    unknown = (
+        connection.execute(
+            text("select distinct status from execution_attempt where status not in :values").bindparams(
+                bindparam("values", recognised, expanding=True)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    if unknown:
+        raise RuntimeError(f"execution_attempt.status carries unrecognised spellings: {sorted(unknown)}")
+
+    rewritten = 0
+    for name, value in sorted(name_to_value.items()):
+        if name == value:
+            continue
+        count = connection.execute(
+            text("select count(*) from execution_attempt where status = :name"),
+            {"name": name},
+        ).scalar_one()
+        if not count:
+            continue
+        print(f"  {name} -> {value}: {count} row(s)")
+        rewritten += count
+        if dry_run:
+            continue
+        connection.execute(
+            text("update execution_attempt set status = :value where status = :name"),
+            {"name": name, "value": value},
+        )
+
+    print(f"  rewritten={rewritten}")
+    return rewritten
+
+
+def strip_job_source_columns(connection: Connection, *, dry_run: bool) -> int:
+    """Step 4: drop the evidence columns from ``job_source``.
+
+    Uses ``ALTER TABLE ... DROP COLUMN``, supported by SQLite 3.35+ and by
+    PostgreSQL. Idempotent: a column that is already gone is skipped.
+    """
+    inspector = sqlalchemy_inspect(connection)
+    present = {column["name"] for column in inspector.get_columns("job_source")}
+    targets = [name for name in JOB_SOURCE_DROPPED_COLUMNS if name in present]
+    if not targets:
+        print("  all evidence columns already dropped")
+        return 0
+
+    print(f"  dropping {len(targets)} column(s): {', '.join(targets)}")
+    if dry_run:
+        return len(targets)
+
+    for name in targets:
+        connection.execute(text(f'alter table "job_source" drop column "{name}"'))
+    return len(targets)
+
+
+def migrate(*, settings: Settings, artifact_dir: Path, dry_run: bool, strict_counts: bool) -> None:
+    """Apply every V4.7 migration step in order."""
+    engine = create_engine(_sync_url(settings))
+    try:
+        with engine.begin() as connection:
+            _preflight(connection, strict=strict_counts)
+
+            print("\nStep 1: rotate stored images")
+            rotate_stored_images(connection, dry_run=dry_run)
+
+            print(f"\nStep 2: drop {ARTIFACT_TABLE}")
+            drop_processing_artifacts(connection, artifact_dir, dry_run=dry_run)
+
+            print("\nStep 3: normalize execution_attempt.status spelling")
+            normalize_attempt_status(connection, dry_run=dry_run)
+
+            print("\nStep 4: strip evidence columns from job_source")
+            strip_job_source_columns(connection, dry_run=dry_run)
+    finally:
+        engine.dispose()
+
+    if dry_run:
+        print("\nDry run: nothing was written.")
+    else:
+        print("\nDone.")
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--dry-run", action="store_true", help="Report what would change without writing")
+    parser.add_argument(
+        "--allow-count-mismatch",
+        action="store_true",
+        help="Warn instead of aborting when row counts differ from the recorded V4.6 snapshot",
+    )
+    parser.add_argument(
+        "--artifact-dir",
+        type=Path,
+        default=DEFAULT_ARTIFACT_DIR,
+        help="Directory that held external artifact files before V4.7",
+    )
+    args = parser.parse_args(argv)
+
+    settings = get_settings()
+    print(f"Target: {_sync_url(settings)}")
+
+    migrate(
+        settings=settings,
+        artifact_dir=args.artifact_dir,
+        dry_run=args.dry_run,
+        strict_counts=not args.allow_count_mismatch,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())