From de18c2e9da2d8a974357f952c57057c5650934ab Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sun, 23 Aug 2026 18:02:04 -0500 Subject: [PATCH] Fix workflow commit atomicity, error path leak, and UI error boundary 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 <223556219+Copilot@users.noreply.github.com> --- .github/instructions/services.instructions.md | 13 ++ docs/error_handling.md | 15 ++ src/transcription/errors.py | 46 +++++- src/transcription/services/workflows.py | 71 +++++++-- src/transcription/ui/pages/home_page.py | 9 +- src/transcription/ui/pages/people_page.py | 12 +- tests/integration/test_pipeline_atomicity.py | 144 ++++++++++++++++++ tests/test_errors.py | 30 +++- tests/test_ui_boundaries.py | 39 +++++ 9 files changed, 356 insertions(+), 23 deletions(-) create mode 100644 tests/integration/test_pipeline_atomicity.py diff --git a/.github/instructions/services.instructions.md b/.github/instructions/services.instructions.md index acd85c5..99ff90f 100644 --- a/.github/instructions/services.instructions.md +++ b/.github/instructions/services.instructions.md @@ -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. diff --git a/docs/error_handling.md b/docs/error_handling.md index 46b288c..8bf3c35 100644 --- a/docs/error_handling.md +++ b/docs/error_handling.md @@ -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. diff --git a/src/transcription/errors.py b/src/transcription/errors.py index 034650f..b6c0eb1 100644 --- a/src/transcription/errors.py +++ b/src/transcription/errors.py @@ -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) diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 67fde77..37195a7 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -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 diff --git a/src/transcription/ui/pages/home_page.py b/src/transcription/ui/pages/home_page.py index 4ae47db..a96c51b 100644 --- a/src/transcription/ui/pages/home_page.py +++ b/src/transcription/ui/pages/home_page.py @@ -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") diff --git a/src/transcription/ui/pages/people_page.py b/src/transcription/ui/pages/people_page.py index f9ba550..577078b 100644 --- a/src/transcription/ui/pages/people_page.py +++ b/src/transcription/ui/pages/people_page.py @@ -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") diff --git a/tests/integration/test_pipeline_atomicity.py b/tests/integration/test_pipeline_atomicity.py new file mode 100644 index 0000000..dadf7af --- /dev/null +++ b/tests/integration/test_pipeline_atomicity.py @@ -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." diff --git a/tests/test_errors.py b/tests/test_errors.py index c72d93b..4ab7395 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -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 = { diff --git a/tests/test_ui_boundaries.py b/tests/test_ui_boundaries.py index 886d8d4..45c6772 100644 --- a/tests/test_ui_boundaries.py +++ b/tests/test_ui_boundaries.py @@ -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}" + )