Fix workflow commit atomicity, error path leak, and UI error boundary
Quality Gate / gate (push) Failing after 48s

Phase 1 of docs/reviews/2026-08-23-code-review.md.

HIGH-01: process_queued_job committed page evidence and the terminal job
status in separate transactions, so a crash between them left a transcript
persisted against a job stuck in PROCESSING that the worker never reclaims.
The final page's write is now deferred into _finalize_batch_outcome so it
shares the terminal transaction. Intermediate pages remain individually
durable, and the terminal commit is shielded against cancellation the same
way per-page writes already were.

HIGH-04: added tests/integration/test_pipeline_atomicity.py covering both
Transaction B and Transaction C. Confirmed failing against the previous
implementation before the fix.

HIGH-03: classify_unexpected_error interpolated the raw exception into
AppError.message, which the UI renders and the API serializes, leaking the
database path from OperationalError. message is now generic. Because message
also feeds format_error_detail, which writes evidence records, the root cause
is preserved on a new internal-only AppError.detail field rather than
discarded.

HIGH-02: replaced 8 hand-rolled ui.notify error calls in home_page and
people_page with error_presenter.show_error, restoring the correlation
error_id, canonical category, and suggestion. Added an AST guard to
test_ui_boundaries.py so pages cannot hand-roll error notifications again.

Docs updated per documentation-sync: the message/detail split in
docs/error_handling.md and the multi-page atomicity rule in
services.instructions.md.

Verification: ruff clean, 381 tests passing, ty unchanged at 10 known
SQLAlchemy descriptor false positives.

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
Jim Lancaster
2026-08-23 18:02:04 -05:00
co-authored by Copilot App
parent 8d3c60fce1
commit de18c2e9da
9 changed files with 356 additions and 23 deletions
@@ -120,6 +120,19 @@ Atomicity rules:
- Terminal state (`TRANSCRIBED` or `FAILED`) and transcript row changes must succeed or roll back together.
- Retry persistence (`QUEUED` + retry increment + error detail) must succeed or roll back together.
### Multi-page batches
These two requirements are in tension for multi-page jobs: each page should be durable as
soon as its provider call returns, but the last page must commit together with the terminal
status. `process_queued_job` resolves it by committing every page except the last one
individually, then deferring the final page's write into `_finalize_batch_outcome` so it
shares the terminal transaction.
Both paths are shielded against cancellation, so the final page is no less durable than the
pages before it. Enforced by `tests/integration/test_pipeline_atomicity.py`; per-page
durability is separately enforced by
`tests/services/test_workflows_reliability.py::TestWorkflowReliability::test_transcribed_page_is_committed_before_next_provider_call_finishes`.
## Contract Alignment
- Treat `docs/` as the active architecture and requirements baseline.
+15
View File
@@ -99,6 +99,21 @@ taxonomy to the six canonical categories at the API/UI envelope boundary.
2. Avoid leaking stack traces or local paths into user-facing message envelopes.
3. Preserve causal exception chains for internal diagnostics.
### Message vs detail split
Rules 1 and 2 pull in opposite directions: evidence records need the root cause, and
user-facing envelopes must not carry it. `AppError` therefore separates the two audiences:
| Field | Audience | Carries root cause | Surfaces |
| --- | --- | --- | --- |
| `message` | User-facing and API-facing | No | `show_error`, `build_error_envelope` |
| `detail` | Internal only | Yes | `format_error_detail` (evidence), logs |
`classify_unexpected_error` builds a generic `message` and puts the exception type and
text on `detail`. Anything rendered to a user or serialized into an API envelope must
read `message`; anything persisted as provenance or logged may read `detail`.
Enforced by `tests/test_errors.py::test_unexpected_error_does_not_leak_filesystem_paths`.
## Operator Recovery Guidance
- **validation/conflict:** correct input or state and retry manually.
+41 -5
View File
@@ -2,12 +2,15 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import UTC
from datetime import datetime
from enum import StrEnum
from uuid import uuid4
logger = logging.getLogger(__name__)
class ErrorCategory(StrEnum):
"""Stable error categories defined by docs/error_handling.md."""
@@ -40,6 +43,7 @@ class AppError(RuntimeError):
suggestion: str = "Retry once. If it persists, review logs and report the error reference id.",
retriable: bool = False,
error_id: str | None = None,
detail: str | None = None,
) -> None:
super().__init__(message)
self.message = message
@@ -47,6 +51,10 @@ class AppError(RuntimeError):
self.suggestion = suggestion
self.retriable = retriable
self.error_id = error_id or new_error_id()
# Internal-only diagnostic text. Persisted to evidence and logs, never rendered
# to users or serialized into API envelopes, because it may embed local
# filesystem paths and other infrastructure detail.
self.detail = detail
@dataclass(frozen=True)
@@ -89,15 +97,43 @@ def build_error_envelope(error: AppError) -> ErrorEnvelope:
def classify_unexpected_error(exc: Exception, *, operation: str) -> AppError:
"""Normalize unknown exceptions into internal_unexpected_error."""
return AppError(
f"Unexpected error during {operation}: {exc}",
"""Normalize unknown exceptions into internal_unexpected_error.
The exception text is deliberately excluded from ``message``. ``AppError.message``
is rendered directly to users by the UI error presenter and is serialized into API
responses by :func:`build_error_envelope`, and unexpected exceptions routinely embed
local filesystem paths (SQLAlchemy ``OperationalError`` carries the database path,
``OSError`` carries the storage root). Leaking those is forbidden by
``.github/instructions/error-handling.instructions.md``.
The detail is preserved on ``AppError.detail`` and logged against ``error_id``. That
keeps the root cause in evidence records and operator logs, which are internal, while
keeping it out of user-facing and API-facing text.
"""
error = AppError(
f"Unexpected error during {operation}.",
category=ErrorCategory.INTERNAL_UNEXPECTED,
suggestion="Retry once. If it persists, review logs and report the error reference id.",
retriable=False,
detail=f"{type(exc).__name__}: {exc}",
)
logger.error(
"Unexpected error operation=%s error_id=%s",
operation,
error.error_id,
exc_info=exc,
)
return error
def format_error_detail(error: AppError) -> str:
"""Return a compact persisted failure string for transcript.error_detail."""
return f"[{error.category.value}] {error.message} | suggestion={error.suggestion} | error_id={error.error_id}"
"""Return a compact persisted failure string for transcript.error_detail.
This is internal provenance, not user-facing output, so it carries
``AppError.detail`` (the root cause) in addition to the user-safe message.
"""
parts = [f"[{error.category.value}] {error.message}"]
if error.detail:
parts.append(f"detail={error.detail}")
parts.extend((f"suggestion={error.suggestion}", f"error_id={error.error_id}"))
return " | ".join(parts)
+62 -9
View File
@@ -233,6 +233,7 @@ async def process_queued_job( # noqa: PLR0915
successful_pages: list[_SuccessfulPage] = []
failed_pages: list[_FailedPage] = []
pending_final_page: _SuccessfulPage | _FailedPage | None = None
externally_stopped = False
prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings)
@@ -241,7 +242,8 @@ async def process_queued_job( # noqa: PLR0915
# latency on the first attempt of every worker process (review log [55]).
provider = services.sources.provider
for source in sources:
for index, source in enumerate(sources):
is_final_source = index == len(sources) - 1
if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
externally_stopped = True
break
@@ -408,12 +410,18 @@ async def process_queued_job( # noqa: PLR0915
error.category.value,
)
await _persist_page_outcome_durably(
job=job,
services=services,
page=page_outcome,
session=session,
)
if is_final_source:
# The last page's evidence and the job's terminal status must succeed or
# roll back together, so this write is deferred into _finalize_batch_outcome.
# Earlier pages stay individually durable.
pending_final_page = page_outcome
else:
await _persist_page_outcome_durably(
job=job,
services=services,
page=page_outcome,
session=session,
)
if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
externally_stopped = True
@@ -427,11 +435,12 @@ async def process_queued_job( # noqa: PLR0915
elif failed_pages and not successful_pages:
terminal_status = JobStatus.FAILED
updated_job = await _finalize_batch_outcome(
updated_job = await _finalize_batch_outcome_durably(
job=job,
services=services,
status=terminal_status,
session=session,
final_page=pending_final_page,
)
logger.info(
@@ -546,20 +555,64 @@ async def _job_no_longer_processing(
return latest_job.status != JobStatus.PROCESSING
async def _finalize_batch_outcome_durably(
*,
job: Job,
services: ServiceBundle,
status: JobStatus,
session: AsyncSession | None,
final_page: _SuccessfulPage | _FailedPage | None,
) -> Job:
"""Shield the terminal commit so cancellation cannot discard the last provider call.
Mirrors ``_persist_page_outcome_durably``. Without this, deferring the final page
into the terminal transaction would make that page less durable than the pages
before it.
"""
task = asyncio.create_task(
_finalize_batch_outcome(
job=job,
services=services,
status=status,
session=session,
final_page=final_page,
)
)
try:
return await asyncio.shield(task)
except asyncio.CancelledError:
await task
raise
async def _finalize_batch_outcome(
*,
job: Job,
services: ServiceBundle,
status: JobStatus,
session: AsyncSession | None = None,
final_page: _SuccessfulPage | _FailedPage | None = None,
) -> Job:
"""Persist the terminal aggregate status after all page outcomes are durable."""
"""Persist the final page outcome and the terminal aggregate status in one transaction.
``services.instructions.md`` ("Workflow Transaction Boundaries") requires transcript
content and the paired terminal status to succeed or roll back together. Committing
them separately can leave a transcript persisted against a job stuck in PROCESSING,
which the worker never reclaims because it only claims QUEUED rows.
``final_page`` is ``None`` when the batch produced no page outcome to pair with the
status change (no sources, or the batch stopped before the last page).
"""
if session is None:
async with services.jobs._session_scope() as local_session:
if final_page is not None:
await _write_page_outcome(job=job, services=services, page=final_page, session=local_session)
updated_job = await services.jobs.mark_job_status(job.id, status, session=local_session)
await local_session.commit()
return updated_job
if final_page is not None:
await _write_page_outcome(job=job, services=services, page=final_page, session=session)
updated_job = await services.jobs.mark_job_status(job.id, status, session=session)
await session.commit()
return updated_job
+5 -4
View File
@@ -14,6 +14,7 @@ from transcription.services.photos import PhotoError
from transcription.services.photos import PhotosService
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.media_urls import resolve_media_url
from transcription.ui.components.primitives import render_empty_state
from transcription.ui.components.primitives import section_header_row
@@ -209,7 +210,7 @@ def register_page() -> None: # noqa: PLR0915
description=(description_input.value or "").strip() or None,
)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Save failed", operation="homepage.photo.update_description")
return
ui.navigate.to("/homepage/edit")
@@ -217,7 +218,7 @@ def register_page() -> None: # noqa: PLR0915
try:
await photos_service.set_primary(photo_id=current_photo.id)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Update failed", operation="homepage.photo.set_primary")
return
ui.navigate.to("/homepage/edit")
@@ -225,7 +226,7 @@ def register_page() -> None: # noqa: PLR0915
try:
await photos_service.delete_photo(photo_id=current_photo.id)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Delete failed", operation="homepage.photo.delete")
return
ui.navigate.to("/homepage/edit")
@@ -252,7 +253,7 @@ def register_page() -> None: # noqa: PLR0915
file_bytes=payload,
)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Upload failed", operation="homepage.photo.create")
return
ui.notify(f"Uploaded {event.file.name}", type="positive")
ui.navigate.to("/homepage/edit")
+8 -4
View File
@@ -262,7 +262,7 @@ def register_page() -> None: # noqa: PLR0915
file_bytes=payload,
)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Upload failed", operation="people.photo.create")
return
ui.notify("Photo uploaded.", type="positive")
ui.navigate.to(f"/people/{person.id}/photos")
@@ -318,7 +318,11 @@ def register_page() -> None: # noqa: PLR0915
description=(input_control.value or "").strip() or None,
)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(
exc,
title="Save failed",
operation="people.photo.update_description",
)
return
ui.notify("Description saved.", type="positive")
ui.navigate.to(f"/people/{person.id}/photos")
@@ -327,7 +331,7 @@ def register_page() -> None: # noqa: PLR0915
try:
await photos_service.set_primary(photo_id=photo_id)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Update failed", operation="people.photo.set_primary")
return
ui.notify("Primary photo updated.", type="positive")
ui.navigate.to(f"/people/{person.id}/photos")
@@ -336,7 +340,7 @@ def register_page() -> None: # noqa: PLR0915
try:
await photos_service.delete_photo(photo_id=photo_id)
except PhotoError as exc:
ui.notify(str(exc), type="negative")
show_error(exc, title="Delete failed", operation="people.photo.delete")
return
ui.notify("Photo deleted.", type="positive")
ui.navigate.to(f"/people/{person.id}/photos")
@@ -0,0 +1,144 @@
"""Atomicity guards for the workflow transaction boundaries.
`.github/instructions/services.instructions.md` ("Workflow Transaction Boundaries")
requires that transcript content and the paired terminal/retry job status change
succeed or roll back together. The existing pipeline tests assert the happy-path
end state, which passes identically whether those writes shared one commit or used
two, so a split-commit regression was invisible to the suite.
These tests inject a fault between the paired writes. They fail if the pair is
split across separate transactions.
"""
import contextlib
from pathlib import Path
from uuid import uuid4
import pytest
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobSource
from transcription.db.models import JobStatus
from transcription.db.models import Source
from transcription.providers.base import TranscriptionResult
from transcription.services import ServiceBundle
from transcription.services.workflows import advance_job
from transcription.services.workflows import process_next_queued_job
FIXTURE_IMAGE = Path("tests/fixtures/images/real/Book Two - page 02.jpg")
async def _seed_single_page_job(services: ServiceBundle) -> tuple[Job, Document]:
"""Create a QUEUED job with exactly one linked source."""
async with services.jobs._session_scope() as session:
document = Document(id=uuid4(), name="atomicity-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="page-1.jpg",
filename="page-1.jpg",
file_path=str(FIXTURE_IMAGE),
file_hash="a" * 64,
file_size_bytes=1,
)
session.add(source)
await session.flush()
session.add(JobSource(job_id=job.id, source_id=source.id))
await session.commit()
return job, document
@pytest.mark.integration
class TestWorkflowTransactionAtomicity:
"""Verify paired transcript and job-status writes share one transaction."""
@pytest.mark.asyncio
async def test_transcript_is_not_committed_when_terminal_status_write_fails(
self,
default_session_factory,
monkeypatch,
):
"""Transaction B: transcript and TRANSCRIBED must roll back together.
A single-page job whose terminal status write fails must not leave the
transcript persisted. If the page outcome commits in its own transaction,
the attempt survives while the job never reaches TRANSCRIBED, which is the
stranded-job state the contract exists to prevent.
"""
services = ServiceBundle.from_session_factory(default_session_factory)
job, _document = await _seed_single_page_job(services)
job_id = job.id
async def _succeeds(*args, **kwargs):
_ = (args, kwargs)
return TranscriptionResult(text="atomic page text", provider="fixture", model="model")
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds)
original_mark = services.jobs.mark_job_status
terminal_statuses = {JobStatus.TRANSCRIBED, JobStatus.PARTIAL_SUCCESS, JobStatus.FAILED}
async def _fail_terminal_write(job_id_arg, status, session=None):
if status in terminal_statuses:
raise RuntimeError("injected fault between transcript and terminal status writes")
return await original_mark(job_id_arg, status, session=session)
monkeypatch.setattr(services.jobs, "mark_job_status", _fail_terminal_write)
assert await process_next_queued_job(services=services) is True
attempts = await services.evidence.list_execution_attempts(job_id=job_id)
transcribed = [attempt for attempt in attempts if attempt.raw_transcription]
assert transcribed == [], (
"Transcript was committed even though the paired terminal status write failed. "
"The page outcome and the terminal status must share one transaction."
)
@pytest.mark.asyncio
async def test_retry_status_and_count_are_not_persisted_when_finalization_fails(
self,
default_session_factory,
default_settings,
monkeypatch,
):
"""Transaction C: QUEUED transition and retry increment must roll back together.
A fault while finalizing the retry write must leave the job exactly as it
was. A split write would requeue the job without incrementing retry_count,
letting it retry without bound.
"""
services = ServiceBundle.from_session_factory(default_session_factory)
job, _document = await _seed_single_page_job(services)
job_id = job.id
async with services.jobs._session_scope() as session:
failed_job = await session.get(Job, job_id)
assert failed_job is not None
failed_job.status = JobStatus.FAILED
await session.commit()
retry_settings = default_settings.model_copy(update={"worker_max_retries": 1})
async def _boom(**kwargs):
_ = kwargs
raise RuntimeError("injected fault during retry finalization")
monkeypatch.setattr(services.jobs, "_finalize", _boom)
reloaded = await services.jobs.read_job(job_id=job_id)
with contextlib.suppress(RuntimeError):
await advance_job(job=reloaded, services=services, settings=retry_settings)
async with services.jobs._session_scope() as session:
final = await session.get(Job, job_id)
assert final is not None
assert final.status == JobStatus.FAILED, "Job was requeued despite the retry write failing."
assert final.retry_count == 0, "retry_count was persisted despite the retry write failing."
+29 -1
View File
@@ -7,6 +7,7 @@ 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 format_error_detail
from transcription.errors import new_error_id
@@ -45,10 +46,37 @@ class TestAppErrorHelpers:
assert isinstance(err, AppError)
assert err.category == ErrorCategory.INTERNAL_UNEXPECTED
assert "unit.test" in err.message
assert "boom" in err.message
# The raw exception text must stay out of the user-facing message: it is rendered
# by the UI presenter and serialized into API envelopes, and unexpected exceptions
# routinely embed local filesystem paths.
assert "boom" not in err.message
assert err.suggestion
assert err.error_id
def test_unexpected_error_does_not_leak_filesystem_paths(self):
"""User-facing and API-facing text must not carry local filesystem paths.
`.github/instructions/error-handling.instructions.md` forbids leaking local
filesystem paths in user-facing output. A SQLAlchemy OperationalError embeds the
database path and an OSError embeds the storage root, so the generic catch-all
path is where that leak would occur. The cause is retained on `detail`, which is
internal-only, so evidence records and logs keep full diagnostic value.
"""
secret_path = r"C:\Github\transcription\data\transcription.db"
exc = OSError(f"unable to open database file: {secret_path}")
err = classify_unexpected_error(exc, operation="worker.process_job")
envelope = build_error_envelope(err)
assert secret_path not in err.message
assert secret_path not in envelope.message
assert secret_path not in err.suggestion
# Internal surfaces keep the root cause.
assert err.detail is not None
assert secret_path in err.detail
assert secret_path in format_error_detail(err)
def test_envelope_categories_use_canonical_contract_values(self):
"""API/UI envelope categories are normalized to canonical short identifiers."""
expected_mapping = {
+39
View File
@@ -113,3 +113,42 @@ def test_only_the_designated_owners_construct_a_raw_table():
if _calls_ui_table(ast.parse(path.read_text(encoding="utf-8")))
)
assert set(offenders) == TABLE_OWNERS
def _notifies_negative(tree: ast.Module) -> bool:
"""Return True if the module calls ``ui.notify(..., type="negative")``."""
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if not (
isinstance(func, ast.Attribute)
and func.attr == "notify"
and isinstance(func.value, ast.Name)
and func.value.id == "ui"
):
continue
for keyword in node.keywords:
if (
keyword.arg == "type"
and isinstance(keyword.value, ast.Constant)
and keyword.value.value == "negative"
):
return True
return False
def test_no_page_hand_rolls_error_notifications():
"""HIGH-02: `ui.instructions.md:42` routes all error display through error_presenter.
Hand-rolled ``ui.notify(str(exc), type="negative")`` discards the correlation
``error_id``, the canonical category, and the actionable suggestion that
``show_error`` renders, leaving the user with nothing to report. Eight such sites
existed in ``home_page`` and ``people_page``; this keeps them from returning.
"""
offenders = sorted(
path.stem for path in _page_paths() if _notifies_negative(ast.parse(path.read_text(encoding="utf-8")))
)
assert offenders == [], (
f"Pages must render errors via error_presenter.show_error, not ui.notify: {offenders}"
)