Phase 5: contain worker faults instead of discarding the classification

Review log [8]. classify_unexpected_error already returned retriable=False and
the verdict was logged and then thrown away. Measured across src/: retriable
was assigned in 9 places and read in none.

The plan asks for a test that a programming error "does not silently retry".
Probing with an injected AttributeError showed that is not what happens, and
the two real failure modes need different fixes.

Mode A, raised after the claim commits (inside advance_job): raised exactly
once, job left at PROCESSING, retry_count 0, never re-claimed, because
claim_next_queued_job filters status == QUEUED. A permanently stranded job
with one swallowed log line, not a retry. advance_job's PROCESSING branch,
commented "Recover mid-flight jobs", is unreachable from the worker for the
same reason.

Mode B, raised before or during the claim: 20 raises in 1.2s, an unbounded hot
spin at the poll interval. It never reaches the per-job retry machinery, so
WORKER_MAX_RETRIES does not cap it and the plan's 60s worst case understates
this path.

services/workflows.py
  _advance_job_with_containment wraps advance_job. Any escaping exception is
  classified and the job driven to terminal FAILED, which is visible in the UI
  and resubmittable. The caller session is rolled back first and the terminal
  write runs in its own transaction, so it stays atomic even when the failure
  left that session dirty (plan task 3). The loop continues, so one poison job
  cannot halt transcription for every other job.

worker.py
  handle_worker_exceptions re-raises non-retriable faults rather than
  suppressing them; retriable ones are still suppressed so transient
  conditions do not stop work. run_worker_loop catches that, logs CRITICAL and
  returns cleanly. Returning rather than propagating matters: the exception
  would otherwise surface only at app shutdown, through the wait_for in
  worker_consumer_lifespan.

tests
  test_run_worker_loop_survives_process_next_exception asserted the loop
  SURVIVES a RuntimeError and continues, which is the Mode B defect written
  down as an expectation. Replaced by
  test_run_worker_loop_stops_on_non_retriable_exception, with a new
  test_run_worker_loop_survives_retriable_exception so suppression of genuinely
  transient faults stays covered, and
  test_error_after_claim_fails_the_job_instead_of_stranding_it for Mode A.

  All three were verified to fail on pre-fix code. The Mode B guard fails by
  timing out, which is the infinite spin made visible.

Verified: 295 passed, 4 skipped, 0 ruff, 0 ty.

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

tests/services/test_workflows_reliability.py

@@ -178,6 +178,63 @@ 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_error_after_claim_fails_the_job_instead_of_stranding_it(
+        self,
+        default_session_factory,
+        monkeypatch,
+    ):
+        """A non-retriable fault after the claim drives the job terminal, not stuck.
+
+        Regression guard for review log [8]. The claim commits PROCESSING before any
+        provider work, and claim_next_queued_job only ever selects QUEUED, so an
+        exception escaping advance_job used to strand the job in PROCESSING forever
+        with one swallowed log line. Measured before the fix: raised once, job left
+        processing, retry_count 0, never re-claimed.
+        """
+        services = ServiceBundle.from_session_factory(default_session_factory)
+        async with services.jobs._session_scope() as session:
+            document = Document(id=uuid4(), name="strand-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="strand.jpg",
+                filename="strand.jpg",
+                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
+                file_hash="e" * 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()
+            job_id = job.id
+
+        async def _succeeds(*args, **kwargs):
+            _ = (args, kwargs)
+            return TranscriptionResult(text="page text", provider="test", model="test-model")
+
+        async def _boom(**kwargs):
+            _ = kwargs
+            raise AttributeError("deliberate programming error")
+
+        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds)
+        monkeypatch.setattr("transcription.services.workflows._finalize_batch_outcome", _boom)
+
+        processed = await workflows_module.process_next_queued_job(services=services)
+
+        assert processed is True
+        async with services.jobs._session_scope() as session:
+            final = await session.get(Job, job_id)
+            assert final is not None
+            # Terminal and resubmittable, rather than stranded in PROCESSING.
+            assert final.status == JobStatus.FAILED
+
     @pytest.mark.asyncio
     async def test_completed_page_is_committed_before_next_provider_call_finishes(
         self,

tests/test_worker.py

@@ -4,6 +4,8 @@ from typing import cast
 
 import pytest
 
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
 from transcription.services import ServiceBundle
 from transcription.services.sources import SourceService
 from transcription.worker import process_next_queued_job
@@ -11,7 +13,38 @@ from transcription.worker import run_worker_loop
 
 
 @pytest.mark.asyncio
-async def test_run_worker_loop_survives_process_next_exception(monkeypatch, caplog):
+async def test_run_worker_loop_stops_on_non_retriable_exception(monkeypatch, caplog):
+    """A programming error before a job is claimed stops the loop instead of spinning.
+
+    Regression guard for review log [8]. This previously spun at the poll interval
+    forever: the fault was classified non-retriable, logged, and then discarded, and
+    it never reached the per-job retry machinery so nothing capped it. Measured at 20
+    iterations in 1.2s before the fix.
+    """
+    calls = 0
+    stop_event = asyncio.Event()
+
+    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
+        nonlocal calls
+        _ = (session, session_factory, services)
+        calls += 1
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
+
+    with caplog.at_level(logging.CRITICAL):
+        await asyncio.wait_for(
+            run_worker_loop(stop_event=stop_event, poll_interval_seconds=0),
+            timeout=5,
+        )
+
+    assert calls == 1
+    assert "Worker loop stopped after a non-retriable error" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog):
+    """A retriable fault is still suppressed so transient conditions do not stop work."""
     calls = 0
     stop_event = asyncio.Event()
 
@@ -20,7 +53,12 @@ async def test_run_worker_loop_survives_process_next_exception(monkeypatch, capl
         _ = (session, session_factory, services)
         calls += 1
         if calls == 1:
-            raise RuntimeError("boom")
+            raise AppError(
+                "transient",
+                category=ErrorCategory.EXTERNAL_PROVIDER,
+                suggestion="retry",
+                retriable=True,
+            )
         stop_event.set()
         return False