gpt-5.3 codex review: Phase 4

docs/phase-4-implementation-task-specs.md

@@ -0,0 +1,253 @@
+# Phase 4: Implementation Task Specifications
+
+## Objective
+
+Convert the canonical V4 baseline into executable, worker-ready implementation tasks that close remaining drift risk and harden enforcement.
+
+Canonical authority for all tasks:
+
+- `docs/ver4/*`
+- `.github/instructions/services.instructions.md`
+- `.github/instructions/ui.instructions.md`
+- `.github/instructions/error-handling.instructions.md`
+- `.github/skills/python-code-reviewer/skill.md`
+- `.github/skills/evidence-provenance-auditor/skill.md`
+
+---
+
+## Dependency-Ordered Task Graph
+
+| Task ID | Title | Depends On | Priority |
+| :--- | :--- | :--- | :--- |
+| P4-01 | Meta-contract regression guards | - | P0 |
+| P4-02 | Error taxonomy and translation conformance | P4-01 | P0 |
+| P4-03 | Evidence append-only write-path hardening | P4-01 | P0 |
+| P4-04 | Job/Page lifecycle consistency hardening | P4-02, P4-03 | P0 |
+| P4-05 | Eager-loading and `lazy="raise"` conformance sweep | P4-01 | P1 |
+| P4-06 | Media route/path safety enforcement sweep | P4-01 | P1 |
+| P4-07 | Duration semantics split for telemetry correctness | P4-04 | P1 |
+| P4-08 | Reviewer/instruction enforcement tests in CI path | P4-01..P4-07 | P1 |
+
+---
+
+## Task Specifications
+
+## P4-01 — Meta-contract regression guards
+
+**Goal:** Add deterministic tests preventing drift from canonical docs/instructions/skills contracts.
+
+**Scope:**
+
+- `tests/` (new focused tests)
+- optionally small helper in `tests/conftest.py` if needed
+
+**Required changes:**
+
+1. Add a test that fails if active docs/instructions/skills reference removed V4 revision paths.
+2. Add a test that asserts canonical authority references exist where required (services/ui/error-handling instructions and reviewer skills).
+3. Keep checks text-based and stable; avoid brittle full-document snapshots.
+
+**Acceptance criteria:**
+
+- Test suite fails on any reintroduction of legacy-path authority references.
+- Test suite passes with current baseline.
+
+**Validation command:** `uv run pytest tests -k "contract or instruction or skill"`
+
+---
+
+## P4-02 — Error taxonomy and translation conformance
+
+**Goal:** Ensure service/API/UI error translation matches canonical categories and boundaries.
+
+**Scope:**
+
+- `src/transcription/services/**/*.py`
+- `src/transcription/api/**/*.py`
+- `src/transcription/ui/**/*.py`
+- related tests under `tests/services`, `tests/api`, `tests/ui`
+
+**Required changes:**
+
+1. Normalize service-edge exceptions to canonical categories:
+   - `validation`, `not_found`, `conflict`, `external`, `timeout`, `internal`.
+2. Ensure causal chain preservation (`raise ... from ...`) on translation.
+3. Ensure UI/API emits user-safe envelopes/messages without stack/path/secret leakage.
+4. Add/adjust tests for category mapping and envelope shape.
+
+**Acceptance criteria:**
+
+- No ad hoc category values in user/API-facing error envelopes.
+- Translation boundaries are deterministic and test-covered.
+
+**Validation command:** `uv run pytest tests/services tests/api tests/ui -k "error or envelope or category"`
+
+---
+
+## P4-03 — Evidence append-only write-path hardening
+
+**Goal:** Guarantee each provider call appends new `ExecutionAttempt` evidence and never rewrites attempt history.
+
+**Scope:**
+
+- `src/transcription/services/sources.py`
+- `src/transcription/services/workflows.py`
+- `src/transcription/services/evidence.py`
+- `src/transcription/db/models.py` (only if required for bug fix)
+- tests under `tests/services`
+
+**Required changes:**
+
+1. Audit all attempt-write paths for accidental update/overwrite behavior.
+2. Enforce append-only semantics for retries and reprocessing.
+3. Keep `Source.raw_transcription` and preferred pointers as projection-only mutation surfaces.
+4. Add/strengthen tests proving historical attempts are unchanged across retries/promotions.
+
+**Acceptance criteria:**
+
+- Reprocessing/retry produces new attempts, never history rewrite.
+- Projection changes do not mutate prior attempt evidence.
+
+**Validation command:** `uv run pytest tests/services -k "attempt or evidence or retry or promotion"`
+
+---
+
+## P4-04 — Job/Page lifecycle consistency hardening
+
+**Goal:** Enforce coherent status transitions and cancellation semantics across `Job` and `JobSource`.
+
+**Scope:**
+
+- `src/transcription/services/jobs.py`
+- `src/transcription/services/workflows.py`
+- related tests under `tests/services`
+
+**Required changes:**
+
+1. Verify terminal `Job.status` resolution from page outcomes only:
+   - `transcribed`, `partial_success`, `failed`.
+2. Ensure cancellation transitions remaining `pending` pages to `cancelled`.
+3. Ensure lifecycle language and emitted behavior never reintroduce legacy `completed` semantics.
+4. Add/adjust reliability tests for mixed outcomes and cancel/retry edges.
+
+**Acceptance criteria:**
+
+- Job and page-level transitions remain consistent under success/failure/partial/cancel paths.
+- No legacy terminal vocabulary in behavior contracts or emitted statuses.
+
+**Validation command:** `uv run pytest tests/services/test_workflows_reliability.py tests/services -k "job or cancel or partial_success"`
+
+---
+
+## P4-05 — Eager-loading and `lazy="raise"` conformance sweep
+
+**Goal:** Eliminate hidden lazy-load access in service/UI read paths.
+
+**Scope:**
+
+- `src/transcription/services/**/*.py`
+- `src/transcription/ui/pages/**/*.py`
+- `src/transcription/ui/components/**/*.py`
+
+**Required changes:**
+
+1. Audit relationship access sites and ensure explicit eager-loads where needed.
+2. Fix any read-path assumptions that rely on implicit lazy loading.
+3. Add targeted tests for high-risk rendering/read paths.
+
+**Acceptance criteria:**
+
+- Read paths that touch relationships function with `lazy="raise"` constraints intact.
+
+**Validation command:** `uv run pytest tests/services tests/ui -k "boundary or lazy or load"`
+
+---
+
+## P4-06 — Media route/path safety enforcement sweep
+
+**Goal:** Ensure all UI/print media flows use approved record-validated or controlled resolver patterns.
+
+**Scope:**
+
+- `src/transcription/api/v4_print.py`
+- `src/transcription/ui/components/media_urls.py`
+- `src/transcription/ui/pages/**/*.py`
+- `src/transcription/ui/components/**/*.py`
+- tests in `tests/ui` and `tests/api` where applicable
+
+**Required changes:**
+
+1. Remove any direct filesystem path exposure to clients.
+2. Ensure print/export media access is record-validated.
+3. Ensure UI media URL construction uses resolver helper only.
+4. Add tests that fail on unsafe URL/path construction patterns.
+
+**Acceptance criteria:**
+
+- No `file://` usage or raw absolute-path emission in UI/API responses.
+- Media links resolve through approved paths only.
+
+**Validation command:** `uv run pytest tests/ui tests/api -k "media or print or path"`
+
+---
+
+## P4-07 — Duration semantics split for telemetry correctness
+
+**Goal:** Separate provider-call latency from end-to-end processing duration so model rollups are not misleading.
+
+**Scope:**
+
+- `src/transcription/services/workflows.py`
+- `src/transcription/db/models.py` and migration surface if schema needs extension
+- `docs/ver4/schema_v4.md` (same change if model fields evolve)
+- tests in `tests/services`
+
+**Required changes:**
+
+1. Introduce explicit timing fields or calculation paths that distinguish:
+   - provider call duration
+   - full processing duration
+2. Keep existing invariants and backward compatibility behavior explicit.
+3. Add tests covering success, timeout, and failure timing writes.
+4. Update schema docs if persistence contract changes.
+
+**Acceptance criteria:**
+
+- Telemetry consumers can compute provider performance without preprocessing/commit skew.
+- Timing semantics are test-backed and documented.
+
+**Validation command:** `uv run pytest tests/services -k "duration or timeout or workflow"`
+
+---
+
+## P4-08 — Enforcement coverage in CI path
+
+**Goal:** Ensure critical invariant checks run in the normal validation path.
+
+**Scope:**
+
+- existing test organization and runner configs only
+
+**Required changes:**
+
+1. Ensure boundary/provenance/contract tests are discoverable by default `pytest` runs.
+2. Add grouping markers only if already consistent with project test conventions.
+3. Avoid introducing new tooling; reuse existing test stack.
+
+**Acceptance criteria:**
+
+- A standard repo test run exercises meta-contract, boundary, and provenance guards.
+
+**Validation command:** `uv run pytest`
+
+---
+
+## Worker Handoff Contract (apply to every task)
+
+Each worker delivery must include:
+
+1. Files changed.
+2. Behavior change summary.
+3. Why the change satisfies the invariant/contract.
+4. Targeted test command(s) and result summary.
+5. Explicit note if `docs/ver4/schema_v4.md` was updated due to model/persistence changes.

docs/ver4/schema_v4.md

@@ -198,7 +198,7 @@ Index:
 | `router_request_id` | `str \| None` | optional |
 | `router_generation_id` | `str \| None` | optional |
 | `sdk_response_snapshot` | `dict[str, JsonValue] \| None` | JSONBCompat |
-| `normalized_metadata` | `dict[str, JsonValue] \| None` | JSONBCompat |
+| `normalized_metadata` | `dict[str, JsonValue] \| None` | JSONBCompat; may include app-namespaced `processing_timing` (`provider_call_duration_ms`, `processing_duration_ms`) |
 | `software_context` | `dict[str, JsonValue] \| None` | JSONBCompat |
 | `raw_transcription` | `str \| None` | optional |
 | `error_category` | `str \| None` | optional |

src/transcription/errors.py

@@ -59,11 +59,28 @@ class ErrorEnvelope:
     timestamp: str
 
 
+def canonical_error_category(error: AppError) -> str:
+    """Map internal categories to canonical API/UI envelope categories."""
+    match error.category:
+        case ErrorCategory.VALIDATION | ErrorCategory.USER_INPUT:
+            return "validation"
+        case ErrorCategory.NOT_FOUND:
+            return "not_found"
+        case ErrorCategory.CONFLICT:
+            return "conflict"
+        case ErrorCategory.EXTERNAL_PROVIDER:
+            return "external"
+        case ErrorCategory.INFRA_TRANSIENT:
+            return "timeout"
+        case _:
+            return "internal"
+
+
 def build_error_envelope(error: AppError) -> ErrorEnvelope:
     """Build an API-safe response envelope from an AppError."""
     return ErrorEnvelope(
         error_id=error.error_id,
-        category=error.category.value,
+        category=canonical_error_category(error),
         message=error.message,
         suggestion=error.suggestion,
         timestamp=datetime.now(UTC).isoformat(),

src/transcription/services/sources.py

@@ -476,6 +476,7 @@ class SourceService(ServiceBase):
         model: str | None = None,
         request_manifest: RequestManifest | None = None,
         quality_warnings: dict[str, JsonValue] | None = None,
+        timing_breakdown: dict[str, JsonValue] | None = None,
         transport_evidence: TransportEvidence | None = None,
         failure_phase: str | None = None,
         error_category: str | None = None,
@@ -508,7 +509,12 @@ 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)
+            timing_payload = _validate_json_object(timing_breakdown, field_name="timing_breakdown")
+            attempt_metadata = _merge_attempt_metadata(
+                metadata=metadata_payload,
+                quality_warnings=quality_warnings,
+                timing_breakdown=timing_payload,
+            )
 
             existing_job_source = await _session.exec(
                 select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
@@ -657,20 +663,24 @@ def _validate_transcription_metadata(
     return validated.as_json_object()
 
 
-def _merge_quality_warnings(
+def _merge_attempt_metadata(
     metadata: dict[str, JsonValue] | None,
+    *,
     quality_warnings: dict[str, JsonValue] | None,
+    timing_breakdown: dict[str, JsonValue] | None,
 ) -> dict[str, JsonValue] | None:
-    """Attach app-computed quality warnings to provider-normalized metadata.
+    """Attach app-computed metadata to provider-normalized metadata.
 
-    The warnings are derived from the transcription text rather than reported by
+    App-computed values (quality warnings and timing) are namespaced so provider
-    the provider, so they are namespaced under their own key instead of being
+    metadata remains semantically distinct.
-    mixed into the provider's own fields.
     """
-    if quality_warnings is None:
+    if quality_warnings is None and timing_breakdown is None:
         return metadata
     merged: dict[str, JsonValue] = dict(metadata or {})
+    if quality_warnings is not None:
         merged["transcription_quality_warnings"] = quality_warnings
+    if timing_breakdown is not None:
+        merged["processing_timing"] = timing_breakdown
     return merged
 
 

src/transcription/services/workflows.py

@@ -116,7 +116,8 @@ class _SuccessfulPage:
     result: TranscriptionResult
     started_at: datetime
     finished_at: datetime
-    duration_ms: int
+    provider_duration_ms: int
+    processing_duration_ms: int
 
 
 @dataclass(frozen=True)
@@ -125,7 +126,8 @@ class _FailedPage:
     error: AppError
     started_at: datetime
     finished_at: datetime
-    duration_ms: int
+    provider_duration_ms: int
+    processing_duration_ms: int
     request_manifest: RequestManifest | None = None
     transport_evidence: TransportEvidence | None = None
     failure_phase: str | None = None
@@ -273,7 +275,11 @@ async def process_queued_job(  # noqa: PLR0915
                 result=result,
                 started_at=started_at,
                 finished_at=finished_at,
-                duration_ms=max(0, int(elapsed_seconds * 1000)),
+                provider_duration_ms=max(0, int(elapsed_seconds * 1000)),
+                processing_duration_ms=max(
+                    _duration_ms_between(started_at, finished_at),
+                    max(0, int(elapsed_seconds * 1000)),
+                ),
             )
             successful_pages.append(page_outcome)
         except TimeoutError:
@@ -289,10 +295,14 @@ async def process_queued_job(  # noqa: PLR0915
                 error=error,
                 started_at=started_at,
                 finished_at=finished_at,
-                duration_ms=max(
+                provider_duration_ms=max(
                     0,
                     int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
                 ),
+                processing_duration_ms=max(
+                    _duration_ms_between(started_at, finished_at),
+                    max(0, int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000)),
+                ),
                 request_manifest=provider.current_request_manifest,
                 transport_evidence=provider.current_transport_evidence,
                 failure_phase="local_timeout",
@@ -321,10 +331,14 @@ async def process_queued_job(  # noqa: PLR0915
                 error=error,
                 started_at=started_at,
                 finished_at=finished_at,
-                duration_ms=max(
+                provider_duration_ms=max(
                     0,
                     int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
                 ),
+                processing_duration_ms=max(
+                    _duration_ms_between(started_at, finished_at),
+                    max(0, int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000)),
+                ),
                 request_manifest=(
                     result.request_manifest
                     if result is not None
@@ -573,10 +587,14 @@ async def _write_page_outcome(
             model=result.model,
             request_manifest=result.request_manifest,
             quality_warnings=quality_warning_payload(analyze_transcription_quality(result.text)),
+            timing_breakdown={
+                "provider_call_duration_ms": page.provider_duration_ms,
+                "processing_duration_ms": page.processing_duration_ms,
+            },
             transport_evidence=result.transport_evidence,
             started_at=page.started_at,
             finished_at=page.finished_at,
-            duration_ms=page.duration_ms,
+            duration_ms=page.provider_duration_ms,
             session=session,
         )
         return
@@ -591,16 +609,24 @@ async def _write_page_outcome(
         provider=page.provider,
         model=page.model,
         request_manifest=page.request_manifest,
+        timing_breakdown={
+            "provider_call_duration_ms": page.provider_duration_ms,
+            "processing_duration_ms": page.processing_duration_ms,
+        },
         transport_evidence=page.transport_evidence,
         failure_phase=page.failure_phase,
         error_category=page.error.category.value,
         started_at=page.started_at,
         finished_at=page.finished_at,
-        duration_ms=page.duration_ms,
+        duration_ms=page.provider_duration_ms,
         session=session,
     )
 
 
+def _duration_ms_between(started_at: datetime, finished_at: datetime) -> int:
+    return max(0, int((finished_at - started_at).total_seconds() * 1000))
+
+
 def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None:
     text_chars = len(result.text)
     text_lines = _line_count(result.text)

src/transcription/ui/components/media_urls.py

@@ -72,3 +72,36 @@ def resolve_media_url(path: str | None, *, upload_dir: Path, base_url: str) -> s
         return absolute_upload_url(f"/uploads/{quote(normalized)}", base_url=base_url)
 
     return absolute_upload_url(f"/uploads/{quote(path_obj.name)}", base_url=base_url)
+
+
+def public_media_path_label(path: str | None, *, upload_dir: Path) -> str:
+    """Return a safe, non-local path label for UI metadata display."""
+    candidate = (path or "").strip()
+    if not candidate:
+        return "unknown"
+
+    normalized = candidate.replace("\\", "/")
+    lowered = normalized.casefold()
+
+    if normalized.startswith(_UPLOAD_ROUTE_PREFIX):
+        return normalized
+    if lowered.startswith("uploads/"):
+        return f"/{normalized}"
+    if lowered.startswith("data/"):
+        relative = normalized.split("/", 1)[1] if "/" in normalized else ""
+        return f"/uploads/{relative}" if relative else "/uploads"
+    if lowered.startswith(("documents/", "persons/")):
+        return f"/uploads/{normalized}"
+
+    resolved_upload_dir = upload_dir.resolve()
+    path_obj = Path(candidate)
+    if path_obj.is_absolute():
+        absolute_candidate = path_obj.resolve()
+        try:
+            relative = absolute_candidate.relative_to(resolved_upload_dir).as_posix()
+            return f"/uploads/{relative}"
+        except ValueError:
+            # Never expose non-managed absolute filesystem paths.
+            return path_obj.name or "unknown"
+
+    return f"/uploads/{quote(path_obj.name)}"

src/transcription/ui/pages/sources_page.py

@@ -29,6 +29,7 @@ from transcription.ui.components.error_presenter import show_error
 from transcription.ui.components.formatters import parse_uuid
 from transcription.ui.components.guards import parsed_record_id
 from transcription.ui.components.guards import render_record_not_found
+from transcription.ui.components.media_urls import public_media_path_label
 from transcription.ui.components.media_urls import resolve_media_url
 from transcription.ui.components.primitives import destructive_button
 from transcription.ui.components.primitives import render_empty_state
@@ -193,6 +194,7 @@ def register_page() -> None:  # noqa: PLR0915
                     source=source,
                     latest_job_source=latest_job_source,
                     latest_attempt=latest_attempt,
+                    settings=resolve_runtime_settings(request),
                 )
 
     @ui.page("/sources/{source_id}/delete")
@@ -303,9 +305,10 @@ def _render_source_metadata_column(
     source: Source,
     latest_job_source: JobSource | None,
     latest_attempt: LatestExecutionAttempt | None,
+    settings: Settings,
 ) -> None:
     with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
-        _render_source_metadata_zone(source)
+        _render_source_metadata_zone(source, settings=settings)
         _render_source_job_metadata_zone(
             latest_job_source,
             latest_attempt=latest_attempt,
@@ -313,14 +316,14 @@ def _render_source_metadata_column(
         _render_source_revision_logistics_zone(source)
 
 
-def _render_source_metadata_zone(source: Source) -> None:
+def _render_source_metadata_zone(source: Source, *, settings: Settings) -> None:
     with archival_card(title="Source Metadata"):
         metadata_row("Upload Name:", source.upload_name)
         metadata_row("Stored Filename:", source.filename)
         metadata_row("Page Number:", str(source.page_number))
         metadata_row("Document Name:", source.document_name or "Not set")
         metadata_row("Document ID:", str(source.document_id))
-        metadata_row("Stored Path:", source.file_path)
+        metadata_row("Stored Path:", public_media_path_label(source.file_path, upload_dir=settings.upload_dir))
 
 
 def _render_source_job_metadata_zone(

tests/api/test_error_responses.py

@@ -33,13 +33,13 @@ class TestApiErrorResponses:
         assert response.status_code == 400
         payload = response.json()
         assert payload["error_id"] == "abc12345"
-        assert payload["category"] == "validation_error"
+        assert payload["category"] == "validation"
         assert payload["message"] == "Bad upload payload"
         assert payload["suggestion"] == "Upload a non-empty file"
         assert "timestamp" in payload
 
-    def test_unexpected_error_returns_internal_unexpected_envelope(self):
+    def test_unexpected_error_returns_internal_envelope(self):
-        """Unexpected exceptions map to internal_unexpected_error with 500."""
+        """Unexpected exceptions map to canonical internal category with 500."""
         app = FastAPI()
         register_error_handlers(app)
 
@@ -52,6 +52,42 @@ class TestApiErrorResponses:
 
         assert response.status_code == 500
         payload = response.json()
-        assert payload["category"] == "internal_unexpected_error"
+        assert payload["category"] == "internal"
         assert "error_id" in payload
         assert payload["suggestion"]
+
+    @pytest.mark.parametrize(
+        ("category", "expected_status", "expected_envelope_category"),
+        [
+            (ErrorCategory.USER_INPUT, 400, "validation"),
+            (ErrorCategory.NOT_FOUND, 404, "not_found"),
+            (ErrorCategory.CONFLICT, 409, "conflict"),
+            (ErrorCategory.EXTERNAL_PROVIDER, 503, "external"),
+            (ErrorCategory.INFRA_TRANSIENT, 503, "timeout"),
+            (ErrorCategory.INFRA_PERSISTENT, 500, "internal"),
+            (ErrorCategory.INTERNAL_UNEXPECTED, 500, "internal"),
+        ],
+    )
+    def test_app_error_category_mapping_uses_canonical_envelope_taxonomy(
+        self,
+        category: ErrorCategory,
+        expected_status: int,
+        expected_envelope_category: str,
+    ):
+        app = FastAPI()
+        register_error_handlers(app)
+
+        @app.get("/category")
+        def category_route() -> dict[str, str]:
+            raise AppError(
+                "Category test",
+                category=category,
+                suggestion="retry",
+                error_id="cat12345",
+            )
+
+        client = TestClient(app)
+        response = client.get("/category")
+
+        assert response.status_code == expected_status
+        assert response.json()["category"] == expected_envelope_category

tests/api/test_v4_documents.py

@@ -224,4 +224,4 @@ def test_duplicate_document_person_link_returns_conflict_envelope(tmp_path):
 
     assert second.status_code == 409
     payload = second.json()
-    assert payload["category"] == "conflict_error"
+    assert payload["category"] == "conflict"

tests/services/test_v45_candidates.py

@@ -121,3 +121,105 @@ async def test_retranscription_job_locks_source_and_frozen_model(default_session
     assert loaded.model == "vendor/alternate"
     assert loaded.user_prompt == "Transcribe verbatim."
     assert [link.source_id for link in loaded.job_sources] == [source.id]
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_promoting_candidate_does_not_mutate_execution_attempt_history(default_session_factory):
+    settings = Settings(openrouter_api_key="test-key", provider_models=None)
+    services = _services(default_session_factory, settings)
+    source = await _seed_source(services)
+
+    first_job = await services.jobs.create_job(Job(document_id=source.document_id))
+    second_job = await services.jobs.create_job(Job(document_id=source.document_id))
+    await services.sources.create_job_source(JobSource(job_id=first_job.id, source_id=source.id))
+    await services.sources.create_job_source(JobSource(job_id=second_job.id, source_id=source.id))
+
+    await services.sources.update_job_source_transcription(
+        job_id=first_job.id,
+        source_id=source.id,
+        text="baseline",
+        provider="fixture",
+        model="model-a",
+    )
+    await services.sources.update_job_source_transcription(
+        job_id=second_job.id,
+        source_id=source.id,
+        text="candidate",
+        provider="fixture",
+        model="model-b",
+    )
+
+    before = [
+        (
+            attempt.id,
+            attempt.job_id,
+            attempt.source_id,
+            attempt.attempt_number,
+            attempt.status.value,
+            attempt.raw_transcription,
+            attempt.error_category,
+            attempt.error_detail,
+            attempt.failure_phase,
+        )
+        for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
+    ]
+
+    candidate = next(
+        attempt
+        for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
+        if attempt.raw_transcription == "candidate"
+    )
+    await services.evidence.promote_machine_attempt(source_id=source.id, execution_attempt_id=candidate.id)
+
+    after = [
+        (
+            attempt.id,
+            attempt.job_id,
+            attempt.source_id,
+            attempt.attempt_number,
+            attempt.status.value,
+            attempt.raw_transcription,
+            attempt.error_category,
+            attempt.error_detail,
+            attempt.failure_phase,
+        )
+        for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
+    ]
+
+    assert after == before
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_retry_appends_new_attempt_instead_of_rewriting_history(default_session_factory):
+    settings = Settings(openrouter_api_key="test-key", provider_models=None)
+    services = _services(default_session_factory, settings)
+    source = await _seed_source(services)
+
+    job = await services.jobs.create_job(Job(document_id=source.document_id))
+    await services.sources.create_job_source(JobSource(job_id=job.id, source_id=source.id))
+
+    await services.sources.update_job_source_transcription(
+        job_id=job.id,
+        source_id=source.id,
+        text=None,
+        error_detail="timeout",
+        error_category="timeout",
+        provider="fixture",
+        model="model-a",
+        failure_phase="provider_call",
+    )
+    await services.sources.update_job_source_transcription(
+        job_id=job.id,
+        source_id=source.id,
+        text="retry-success",
+        provider="fixture",
+        model="model-a",
+    )
+
+    attempts = list(await services.evidence.list_execution_attempts(source_id=source.id))
+    assert len(attempts) == 2
+    assert [attempt.attempt_number for attempt in attempts] == [1, 2]
+    assert attempts[0].raw_transcription is None
+    assert attempts[1].raw_transcription == "retry-success"

tests/services/test_workflows_reliability.py

@@ -181,6 +181,74 @@ class TestWorkflowReliability:
         assert duration_ms >= int(budget_seconds * 1000 * 0.9)
         assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9)
 
+    @pytest.mark.asyncio
+    async def test_attempt_metadata_persists_provider_and_processing_durations(
+        self,
+        default_session_factory,
+        monkeypatch,
+    ):
+        """Execution metadata records both provider-only and end-to-end durations."""
+        services = ServiceBundle.from_session_factory(default_session_factory)
+        async with services.jobs._session_scope() as session:
+            document = Document(id=uuid4(), name="timing-metadata-doc")
+            session.add(document)
+            await session.flush()
+            job = Job(document_id=document.id, status=JobStatus.QUEUED)
+            session.add(job)
+            await session.flush()
+            source = Source(
+                document_id=document.id,
+                page_number=1,
+                upload_name="timing.jpg",
+                filename="timing.jpg",
+                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
+                file_hash="f" * 64,
+                file_size_bytes=1,
+            )
+            session.add(source)
+            await session.flush()
+            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
+            await session.commit()
+            loaded = await services.jobs.read_job(job_id=job.id, session=session)
+
+        async def _returns_text(*args, **kwargs):
+            _ = (args, kwargs)
+            await asyncio.sleep(0.03)
+            return TranscriptionResult(text="timed output", provider="fixture", model="fixture-model")
+
+        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _returns_text)
+
+        result = await process_queued_job(
+            job=loaded,
+            services=services,
+            settings=Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=2.0),
+        )
+        assert result is not None
+        assert result.status == JobStatus.TRANSCRIBED
+
+        async with services.jobs._session_scope() as session:
+            attempts = (
+                (
+                    await session.exec(
+                        select(ExecutionAttempt).where(
+                            col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
+                        )
+                    )
+                )
+                .all()
+            )
+        assert len(attempts) == 1
+        attempt = attempts[0]
+        timing = (attempt.normalized_metadata or {}).get("processing_timing")
+        assert isinstance(timing, dict)
+        provider_call_ms = timing.get("provider_call_duration_ms")
+        processing_ms = timing.get("processing_duration_ms")
+        assert isinstance(provider_call_ms, int)
+        assert isinstance(processing_ms, int)
+        assert provider_call_ms >= 0
+        assert processing_ms >= provider_call_ms
+        assert attempt.duration_ms == provider_call_ms
+
     @pytest.mark.asyncio
     async def test_error_after_claim_fails_the_job_instead_of_stranding_it(
         self,
@@ -239,7 +307,7 @@ class TestWorkflowReliability:
             assert final.status == JobStatus.FAILED
 
     @pytest.mark.asyncio
-    async def test_completed_page_is_committed_before_next_provider_call_finishes(
+    async def test_transcribed_page_is_committed_before_next_provider_call_finishes(
         self,
         default_session_factory,
         monkeypatch,

tests/test_errors.py

@@ -4,6 +4,8 @@ import pytest
 
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
+from transcription.errors import build_error_envelope
+from transcription.errors import canonical_error_category
 from transcription.errors import classify_unexpected_error
 from transcription.errors import new_error_id
 
@@ -44,3 +46,14 @@ class TestAppErrorHelpers:
         assert "boom" in err.message
         assert err.suggestion
         assert err.error_id
+
+    def test_envelope_categories_use_canonical_contract_values(self):
+        """API/UI envelope categories are normalized to canonical short identifiers."""
+        validation = AppError("x", category=ErrorCategory.USER_INPUT)
+        timeout = AppError("x", category=ErrorCategory.INFRA_TRANSIENT)
+        internal = AppError("x", category=ErrorCategory.INTERNAL_UNEXPECTED)
+
+        assert canonical_error_category(validation) == "validation"
+        assert canonical_error_category(timeout) == "timeout"
+        assert canonical_error_category(internal) == "internal"
+        assert build_error_envelope(validation).category == "validation"

tests/test_media_path_safety.py

@@ -0,0 +1,30 @@
+"""Guardrails for UI/API media path safety contracts."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+UI_ROOT = PROJECT_ROOT / "src" / "transcription" / "ui"
+API_ROOT = PROJECT_ROOT / "src" / "transcription" / "api"
+SOURCES_PAGE = UI_ROOT / "pages" / "sources_page.py"
+
+
+def _python_files(root: Path) -> list[Path]:
+    return sorted(root.rglob("*.py"))
+
+
+def test_no_ui_or_api_python_file_uses_file_scheme_links():
+    violations: list[str] = []
+    for path in [*_python_files(UI_ROOT), *_python_files(API_ROOT)]:
+        text = path.read_text(encoding="utf-8")
+        if "file://" in text:
+            violations.append(str(path.relative_to(PROJECT_ROOT)).replace("\\", "/"))
+    assert violations == []
+
+
+def test_sources_page_does_not_render_raw_file_path_value():
+    text = SOURCES_PAGE.read_text(encoding="utf-8")
+    assert 'metadata_row("Stored Path:", source.file_path)' not in text
+    assert "public_media_path_label(source.file_path" in text

tests/test_meta_contract_guards.py

@@ -0,0 +1,99 @@
+"""Regression guards for canonical documentation and instruction contracts."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+ACTIVE_CONTRACT_FILES = (
+    ".github/instructions/services.instructions.md",
+    ".github/instructions/ui.instructions.md",
+    ".github/instructions/error-handling.instructions.md",
+    ".github/skills/python-code-reviewer/skill.md",
+    ".github/skills/evidence-provenance-auditor/skill.md",
+    "docs/ver4/index_v4.md",
+    "docs/ver4/architecture_v4.md",
+    "docs/ver4/requirements_v4.md",
+    "docs/ver4/schema_v4.md",
+    "docs/ver4/error_handling_v4.md",
+    "docs/ui/README.md",
+)
+
+
+LEGACY_REFERENCE_MARKERS = (
+    "docs-v4x-archive",
+    "docs/ver4/history.md",
+    "docs/ver4.0",
+    "docs/ver4.1",
+    "docs/ver4.2",
+    "docs/ver4.3",
+    "docs/ver4.4",
+    "docs/ver4.5",
+    "docs/ver4.6",
+    "docs/ver4.7",
+)
+
+
+def _read(relative_path: str) -> str:
+    return (PROJECT_ROOT / relative_path).read_text(encoding="utf-8")
+
+
+def test_active_contract_files_are_present():
+    """Guard the guard: ensure all expected authority files are scanned."""
+    missing = [path for path in ACTIVE_CONTRACT_FILES if not (PROJECT_ROOT / path).exists()]
+    assert missing == []
+
+
+def test_no_legacy_authority_references_in_active_contract_files():
+    """Active contracts must not route authority through removed V4 revision trees."""
+    violations: dict[str, list[str]] = {}
+    for relative_path in ACTIVE_CONTRACT_FILES:
+        text = _read(relative_path)
+        found = [marker for marker in LEGACY_REFERENCE_MARKERS if marker in text]
+        if found:
+            violations[relative_path] = found
+    assert violations == {}
+
+
+def test_canonical_authority_references_are_present():
+    """Critical instruction and skill files must keep canonical references explicit."""
+    required_fragments = {
+        ".github/instructions/services.instructions.md": (
+            "docs/ver4/",
+            "./error-handling.instructions.md",
+            "src/transcription/db/models.py",
+            "docs/ver4/schema_v4.md",
+            "append-only",
+        ),
+        ".github/instructions/ui.instructions.md": (
+            "docs/ver4/",
+            "./error-handling.instructions.md",
+            "src/transcription/db/models.py",
+            "docs/ver4/schema_v4.md",
+        ),
+        ".github/instructions/error-handling.instructions.md": (
+            "docs/ver4/error_handling_v4.md",
+            "docs/ver4/requirements_v4.md",
+        ),
+        ".github/skills/python-code-reviewer/skill.md": (
+            "docs/ver4/*",
+            "docs/ver4/schema_v4.md",
+            ".github/instructions/error-handling.instructions.md",
+        ),
+        ".github/skills/evidence-provenance-auditor/skill.md": (
+            "docs/ver4/schema_v4.md",
+            "docs/ver4/requirements_v4.md",
+            "docs/ver4/error_handling_v4.md",
+        ),
+    }
+
+    missing: dict[str, list[str]] = {}
+    for relative_path, fragments in required_fragments.items():
+        text = _read(relative_path)
+        absent = [fragment for fragment in fragments if fragment not in text]
+        if absent:
+            missing[relative_path] = absent
+    assert missing == {}

tests/test_models.py

@@ -157,6 +157,21 @@ class TestJobModel:
 
         assert job.status == JobStatus.TRANSCRIBED
 
+    def test_status_enums_match_v4_lifecycle_contract(self):
+        assert [status.value for status in JobStatus] == [
+            "queued",
+            "processing",
+            "transcribed",
+            "partial_success",
+            "failed",
+        ]
+        assert [status.value for status in JobSourceStatus] == [
+            "pending",
+            "transcribed",
+            "failed",
+            "cancelled",
+        ]
+
 
 class TestSourceModel:
     def test_can_be_created_for_document(self, session):

tests/ui/test_media_urls.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+
+from transcription.ui.components.media_urls import public_media_path_label
+
+
+def test_public_media_path_label_maps_managed_absolute_path_to_upload_route(tmp_path):
+    upload_dir = tmp_path / "uploads"
+    managed_path = upload_dir / "documents" / "abc" / "page.jpg"
+    managed_path.parent.mkdir(parents=True, exist_ok=True)
+    managed_path.write_bytes(b"x")
+
+    label = public_media_path_label(str(managed_path), upload_dir=upload_dir)
+    assert label == "/uploads/documents/abc/page.jpg"
+
+
+def test_public_media_path_label_hides_unmanaged_absolute_path(tmp_path):
+    upload_dir = tmp_path / "uploads"
+    unmanaged_path = tmp_path / "other-root" / "secret" / "page.jpg"
+
+    label = public_media_path_label(str(unmanaged_path), upload_dir=upload_dir)
+    assert label == "page.jpg"
+
+
+def test_public_media_path_label_preserves_upload_relative_route(tmp_path):
+    upload_dir = tmp_path / "uploads"
+    label = public_media_path_label("documents/person-1/photo.png", upload_dir=upload_dir)
+    assert label == "/uploads/documents/person-1/photo.png"
+
+
+def test_public_media_path_label_keeps_existing_upload_route(tmp_path):
+    upload_dir = tmp_path / "uploads"
+    assert public_media_path_label("/uploads/documents/d1/page.png", upload_dir=upload_dir) == (
+        "/uploads/documents/d1/page.png"
+    )