perf: move blocking homepage and hash I/O off loop

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

.env.example

@@ -57,6 +57,10 @@ DATABASE_BACKUP_DIR=./data/backups
 # --- worker reliability ---
 WORKER_MAX_RETRIES=0
 WORKER_PROVIDER_TIMEOUT_SECONDS=30.0
+WORKER_STALE_JOB_SECONDS=30.0
+WORKER_RETRY_BACKOFF_SECONDS=1.0
+WORKER_SHUTDOWN_GRACE_SECONDS=5.0
+WORKER_POLL_INTERVAL_SECONDS=1.0
 WORKER_MIN_TRANSCRIPTION_CHARS=0
 WORKER_MIN_TRANSCRIPTION_LINES=0
 WORKER_FAIL_ON_FINISH_REASON_LENGTH=false

docs/production-runbook.md

@@ -127,3 +127,27 @@ Every suppression must be:
 
 Do not use broad or rationale-free suppressions. If a diagnostic is not a known
 false positive, fix the code instead of suppressing it.
+
+## 8. Worker shutdown budget
+
+Worker shutdown waits for at most:
+
+`WORKER_PROVIDER_TIMEOUT_SECONDS + WORKER_SHUTDOWN_GRACE_SECONDS`
+
+`WORKER_PROVIDER_TIMEOUT_SECONDS` covers an in-flight provider call, and
+`WORKER_SHUTDOWN_GRACE_SECONDS` is extra time for the loop to persist outcomes
+and exit cleanly after the call returns.
+
+Set the container or service termination grace period **above this total**
+budget. If termination grace is shorter, the process may be killed before
+terminal status and evidence writes are finalized.
+
+## 9. Horizontal scaling precondition
+
+Multiple worker replicas can race on execution-attempt numbering for the same
+`(job_id, source_id)` pair. The runtime now retries boundedly on unique-key
+conflicts (`uq_execution_attempt_number`) and surfaces a conflict-domain error
+if retries are exhausted.
+
+Do not deploy additional worker replicas unless this conflict-retry path and its
+tests are present and green in the target build.

src/transcription/app.py

@@ -59,7 +59,10 @@ async def _lifespan(app: FastAPI):
         stop_event, worker_notifier, worker_health = await stack.enter_async_context(
             worker_consumer_lifespan(
                 session_factory=app.state.runtime.session_factory,
-                poll_interval_seconds=1.0,
+                poll_interval_seconds=settings.worker_poll_interval_seconds,
+                shutdown_timeout_seconds=(
+                    settings.worker_provider_timeout_seconds + settings.worker_shutdown_grace_seconds
+                ),
             )
         )
         app.state.worker_stop_event = stop_event
@@ -71,11 +74,11 @@ async def _lifespan(app: FastAPI):
 async def _recover_stale_processing_jobs(app: FastAPI) -> None:
     """Re-queue stale processing jobs at startup.
 
-    Any job left in PROCESSING longer than the configured provider timeout is
-    assumed orphaned and moved back to QUEUED before the worker starts.
+    Any job left in PROCESSING longer than the stale-job threshold is assumed
+    orphaned and moved back to QUEUED before the worker starts.
     """
     settings = app.state.settings
-    stale_before = datetime.now(UTC) - timedelta(seconds=settings.worker_provider_timeout_seconds)
+    stale_before = datetime.now(UTC) - timedelta(seconds=settings.worker_stale_job_seconds)
     recovered = await app.state.services.jobs.requeue_stale_processing_jobs(stale_before=stale_before)
     if recovered > 0:
         logger.warning("Recovered %s stale processing job(s) at startup", recovered)

src/transcription/config.py

@@ -114,6 +114,10 @@ class Settings(BaseSettings):
     # Bounded only from below. Vision transcription of a dense page routinely runs
     # well past twenty seconds, so an upper cap here would silently fail real work.
     worker_provider_timeout_seconds: float = Field(default=30.0, gt=0.0)
+    worker_stale_job_seconds: float = Field(default=30.0, gt=0.0)
+    worker_retry_backoff_seconds: float = Field(default=1.0, ge=0.0)
+    worker_shutdown_grace_seconds: float = Field(default=5.0, ge=0.0)
+    worker_poll_interval_seconds: float = Field(default=1.0, gt=0.0)
     worker_min_transcription_chars: int = Field(default=0, ge=0)
     worker_min_transcription_lines: int = Field(default=0, ge=0)
     worker_fail_on_finish_reason_length: bool = False

src/transcription/services/evidence.py

@@ -43,6 +43,26 @@ class LatestExecutionAttempt:
 class EvidenceService(ServiceBase):
     """Read, project, and export execution attempt evidence."""
 
+    async def read_latest_job_error_category(
+        self,
+        *,
+        job_id: UUID,
+        session: AsyncSession | None = None,
+    ) -> str | None:
+        """Read the latest persisted execution-attempt error category for a job."""
+        async with self._session_scope(session) as _session:
+            query = (
+                select(ExecutionAttempt.error_category)
+                .where(ExecutionAttempt.job_id == job_id)
+                .where(col(ExecutionAttempt.error_category).is_not(None))
+                .order_by(
+                    col(ExecutionAttempt.created_at).desc(),
+                    col(ExecutionAttempt.id).desc(),
+                )
+                .limit(1)
+            )
+            return (await _session.exec(query)).first()
+
     async def read_latest_execution_attempt(
         self,
         *,

src/transcription/services/sources.py

@@ -62,6 +62,7 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_PROMPT_FILE = "transcribe_document.md"
 JSON_OBJECT_ADAPTER = TypeAdapter(dict[str, JsonValue])
+MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES = 3
 
 
 class PromptExecution(BaseModel):
@@ -533,54 +534,72 @@ class SourceService(ServiceBase):
 
             finish_time = finished_at or datetime.now(UTC)
             start_time = started_at or finish_time
-            attempt_number = (
-                await _session.exec(
-                    select(func.max(ExecutionAttempt.attempt_number))
-                    .where(ExecutionAttempt.job_id == job_id)
-                    .where(ExecutionAttempt.source_id == source_id)
-                )
-            ).one()
             transport = transport_evidence or TransportEvidence(response_received=False)
             manifest_payload = request_manifest.model_dump(mode="json") if request_manifest is not None else None
             software_payload = (
                 request_manifest.software.model_dump(mode="json") if request_manifest is not None else None
             )
-            attempt = ExecutionAttempt(
-                job_source_id=job_source.id,
-                job_id=job_id,
-                source_id=source_id,
-                attempt_number=(attempt_number or 0) + 1,
-                status=outcome,
-                provider=provider or job.provider or self.settings.provider.value,
-                model=model or job.model,
-                request_manifest=manifest_payload,
-                request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None,
-                request_manifest_schema_version=(
-                    request_manifest.schema_version if request_manifest is not None else None
-                ),
-                response_received=transport.response_received,
-                transport_status_code=transport.status_code,
-                transport_body=transport.body,
-                transport_content_type=transport.content_type,
-                transport_content_encoding=transport.content_encoding,
-                transport_safe_headers=transport.safe_headers or None,
-                router_request_id=transport.request_id,
-                router_generation_id=transport.generation_id,
-                sdk_response_snapshot=raw_response_payload,
-                normalized_metadata=attempt_metadata,
-                software_context=software_payload,
-                raw_transcription=text,
-                error_category=error_category,
-                error_detail=error_detail,
-                failure_phase=failure_phase,
-                started_at=start_time,
-                finished_at=finish_time,
-                duration_ms=duration_ms
-                if duration_ms is not None
-                else max(0, int((finish_time - start_time).total_seconds() * 1000)),
-            )
-            _session.add(attempt)
-            await _session.flush()
+            attempt: ExecutionAttempt | None = None
+            for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1):
+                latest_attempt_number = (
+                    await _session.exec(
+                        select(func.max(ExecutionAttempt.attempt_number))
+                        .where(ExecutionAttempt.job_id == job_id)
+                        .where(ExecutionAttempt.source_id == source_id)
+                    )
+                ).one()
+                candidate = ExecutionAttempt(
+                    job_source_id=job_source.id,
+                    job_id=job_id,
+                    source_id=source_id,
+                    attempt_number=(latest_attempt_number or 0) + 1,
+                    status=outcome,
+                    provider=provider or job.provider or self.settings.provider.value,
+                    model=model or job.model,
+                    request_manifest=manifest_payload,
+                    request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None,
+                    request_manifest_schema_version=(
+                        request_manifest.schema_version if request_manifest is not None else None
+                    ),
+                    response_received=transport.response_received,
+                    transport_status_code=transport.status_code,
+                    transport_body=transport.body,
+                    transport_content_type=transport.content_type,
+                    transport_content_encoding=transport.content_encoding,
+                    transport_safe_headers=transport.safe_headers or None,
+                    router_request_id=transport.request_id,
+                    router_generation_id=transport.generation_id,
+                    sdk_response_snapshot=raw_response_payload,
+                    normalized_metadata=attempt_metadata,
+                    software_context=software_payload,
+                    raw_transcription=text,
+                    error_category=error_category,
+                    error_detail=error_detail,
+                    failure_phase=failure_phase,
+                    started_at=start_time,
+                    finished_at=finish_time,
+                    duration_ms=duration_ms
+                    if duration_ms is not None
+                    else max(0, int((finish_time - start_time).total_seconds() * 1000)),
+                )
+                try:
+                    async with _session.begin_nested():
+                        _session.add(candidate)
+                        await _session.flush()
+                    attempt = candidate
+                    break
+                except IntegrityError:
+                    logger.warning(
+                        "Execution attempt number conflict job_id=%s source_id=%s retry=%s/%s",
+                        job_id,
+                        source_id,
+                        attempt_retry,
+                        MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
+                    )
+                    continue
+
+            if attempt is None:
+                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id)
 
             if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
                 source.raw_transcription = text
@@ -597,6 +616,17 @@ class SourceService(ServiceBase):
             suggestion="Use the existing job-source link instead of creating a duplicate.",
         )
 
+    @staticmethod
+    def _execution_attempt_conflict(*, job_id: UUID, source_id: UUID) -> TranscriptionError:
+        return TranscriptionError(
+            (
+                f"Failed to allocate an execution attempt number for Source {source_id} in Job {job_id} "
+                "after bounded retries"
+            ),
+            category=ErrorCategory.CONFLICT,
+            suggestion="Retry the transcription. If it repeats, investigate concurrent worker activity.",
+        )
+
     async def upsert_revision_for_source(
         self,
         *,

src/transcription/services/store.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import logging
 from collections.abc import Sequence
@@ -398,6 +399,10 @@ async def store_source_file(
     )
     return StoredSourceFile(
         path=stored_path,
-        file_hash=hashlib.sha256(file_bytes).hexdigest(),
+        file_hash=await asyncio.to_thread(_sha256_hexdigest, file_bytes),
         file_size_bytes=len(file_bytes),
     )
+
+
+def _sha256_hexdigest(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()

src/transcription/services/workflows.py

@@ -40,6 +40,12 @@ from .sources import transcribe_document_image
 
 logger = logging.getLogger(__name__)
 
+_RETRIABLE_FAILED_JOB_ERROR_CATEGORIES = {
+    ErrorCategory.EXTERNAL_PROVIDER.value,
+    ErrorCategory.EXTERNAL_TIMEOUT.value,
+    ErrorCategory.INFRA_TRANSIENT.value,
+}
+
 
 async def create_document_with_people(
     *,
@@ -182,16 +188,33 @@ async def advance_job(
             # Recover mid-flight jobs by continuing the queued processing path.
             return await process_queued_job(job=job, services=services, settings=settings, session=session)
         case JobStatus.FAILED:
-            if job.retry_count < settings.worker_max_retries:
+            latest_error_category = await services.evidence.read_latest_job_error_category(
+                job_id=job.id,
+                session=session,
+            )
+            can_retry = (
+                job.retry_count < settings.worker_max_retries
+                and latest_error_category in _RETRIABLE_FAILED_JOB_ERROR_CATEGORIES
+            )
+            if can_retry:
+                if settings.worker_retry_backoff_seconds > 0:
+                    await asyncio.sleep(settings.worker_retry_backoff_seconds)
                 return await services.jobs.update_job_state(
                     job_id=job.id,
                     status=JobStatus.QUEUED,
                     retry_count_increment=1,
                     session=session,
                 )
+
+            if job.retry_count < settings.worker_max_retries:
+                logger.warning(
+                    "Job %s failed with non-retriable category %s; skipping retry.",
+                    job.id,
+                    latest_error_category or "unknown",
+                )
             else:
                 logger.error("Job %s has failed and reached max retries.", job.id)
-                return
+            return
         case _:
             return
 

src/transcription/ui/pages/home_page.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable
 
 from fastapi import Request
@@ -116,7 +117,7 @@ def _render_homepage_view(*, markdown_text: str, render_image_panel: Callable[[]
             ui.element("div")
 
 
-def _render_homepage_editor(*, render_image_panel, markdown_input, on_upload) -> None:
+def _render_homepage_editor(*, render_image_panel, markdown_input, on_upload, initial_markdown: str) -> None:
     with ui.grid().classes("w-full grid-cols-12 gap-4"):
         with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
             with archival_card(title="Homepage Image"):
@@ -132,7 +133,7 @@ def _render_homepage_editor(*, render_image_panel, markdown_input, on_upload) ->
             markdown_input[0] = (
                 ui.textarea(
                     label="Homepage markdown",
-                    value=read_homepage_markdown(),
+                    value=initial_markdown,
                 )
                 .props("outlined autogrow")
                 .classes("w-full")
@@ -152,6 +153,7 @@ def register_page() -> None:  # noqa: PLR0915
         render_navigation_header(current_path="/homepage")
         photos = await photos_service.list_photos(person_id=None)
         active_index = [0]
+        homepage_markdown = [""]
 
         @ui.refreshable
         def render_image_panel() -> None:
@@ -172,12 +174,20 @@ def register_page() -> None:  # noqa: PLR0915
                     on_click=lambda: ui.navigate.to("/homepage/edit"),
                     icon="edit",
                 ).classes("ui-btn-primary text-xs")
-            _render_homepage_view(
-                markdown_text=read_homepage_markdown().strip(),
-                render_image_panel=render_image_panel,
-            )
 
-    @ui.page("/homepage/edit", title="Edit Homepage")
+            @ui.refreshable
+            def render_home_content() -> None:
+                _render_homepage_view(
+                    markdown_text=homepage_markdown[0].strip(),
+                    render_image_panel=render_image_panel,
+                )
+
+            render_home_content()
+
+        homepage_markdown[0] = await asyncio.to_thread(read_homepage_markdown, settings)
+        render_home_content.refresh()
+
+    @ui.page("/homepage/edit", title="Edit Home Page")
     async def homepage_edit_page(request: Request, session_factory: SessionFactoryDep) -> None:  # noqa: PLR0915
         photos_service = PhotosService(session_factory=session_factory)
         settings = resolve_runtime_settings(request)
@@ -268,7 +278,11 @@ def register_page() -> None:  # noqa: PLR0915
             ui.navigate.to("/homepage/edit")
 
         async def save_homepage() -> None:
-            save_homepage_markdown((markdown_input[0].value if markdown_input[0] is not None else "") or "")
+            await asyncio.to_thread(
+                save_homepage_markdown,
+                (markdown_input[0].value if markdown_input[0] is not None else "") or "",
+                settings,
+            )
             ui.notify("Homepage saved", type="positive")
             ui.navigate.to("/homepage")
 
@@ -283,4 +297,9 @@ def register_page() -> None:  # noqa: PLR0915
                 render_image_panel=render_image_panel,
                 markdown_input=markdown_input,
                 on_upload=on_upload,
+                initial_markdown="",
             )
+
+        loaded_markdown = await asyncio.to_thread(read_homepage_markdown, settings)
+        if markdown_input[0] is not None:
+            markdown_input[0].value = loaded_markdown

src/transcription/ui/pages/settings_page.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 from uuid import UUID
 
@@ -540,8 +541,8 @@ async def _recover_prompt(prompts: PromptStore, name: str) -> None:
 
 
 async def _read_home_page_text(settings: Settings) -> str:
-    return read_homepage_markdown(settings=settings)
+    return await asyncio.to_thread(read_homepage_markdown, settings=settings)
 
 
 async def _write_home_page_text(settings: Settings, markdown_text: str) -> None:
-    save_homepage_markdown(markdown_text, settings=settings)
+    await asyncio.to_thread(save_homepage_markdown, markdown_text, settings=settings)

src/transcription/worker.py

@@ -9,6 +9,9 @@ from contextlib import asynccontextmanager
 from contextlib import contextmanager
 from contextlib import suppress
 from dataclasses import dataclass
+from datetime import UTC
+from datetime import datetime
+from datetime import timedelta
 from typing import Literal
 from typing import Protocol
 from typing import runtime_checkable
@@ -120,6 +123,7 @@ async def worker_consumer_lifespan(
     *,
     session_factory: async_sessionmaker[AsyncSession] | None = None,
     poll_interval_seconds: float = 1.0,
+    shutdown_timeout_seconds: float = 2.0,
 ) -> AsyncGenerator[tuple[asyncio.Event, WorkerNotifier, WorkerHealth]]:
     """Start and stop the worker consumer loop for app lifespan."""
     stop_event = asyncio.Event()
@@ -143,7 +147,7 @@ async def worker_consumer_lifespan(
         stop_event.set()
         worker_notifier.notify()
         try:
-            await asyncio.wait_for(worker_task, timeout=2.0)
+            await asyncio.wait_for(worker_task, timeout=shutdown_timeout_seconds)
         except TimeoutError:
             worker_task.cancel()
             with suppress(asyncio.CancelledError):
@@ -215,6 +219,14 @@ async def run_worker_loop(
                     await asyncio.wait_for(wake_event.wait(), timeout=poll_interval_seconds)
                     wake_event.clear()
 
+            if session_factory is not None:
+                with handle_worker_exceptions(operation="worker.requeue_stale_processing_jobs"):
+                    stale_seconds = services.jobs.settings.worker_stale_job_seconds
+                    stale_before = datetime.now(UTC) - timedelta(seconds=stale_seconds)
+                    recovered = await services.jobs.requeue_stale_processing_jobs(stale_before=stale_before)
+                    if recovered > 0:
+                        logger.warning("Recovered %s stale processing job(s) in worker loop", recovered)
+
             processed_any = False
             while True:
                 with handle_worker_exceptions(operation="worker.process_next_queued_job"):

tests/services/test_v2_crud.py

@@ -2,9 +2,12 @@ from uuid import uuid4
 
 import pytest
 from pydantic import JsonValue
+from sqlalchemy.exc import IntegrityError
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.db.models import Document
 from transcription.db.models import DocumentPerson
+from transcription.db.models import ExecutionAttempt
 from transcription.db.models import Job
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
@@ -369,3 +372,104 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
     assert attempt.attempt.raw_transcription == "provider transcript"
     assert attempt.attempt.normalized_metadata == metadata
     assert attempt.attempt.sdk_response_snapshot == raw_payload
+
+
+@pytest.mark.asyncio
+async def test_update_job_source_transcription_retries_on_execution_attempt_integrity_conflict(
+    default_session_factory,
+    monkeypatch,
+):
+    documents = DocumentService(session_factory=default_session_factory)
+    jobs = JobService(session_factory=default_session_factory)
+    transcriptions = SourceService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="attempt-retry-doc"))
+    job = await jobs.create_job(Job(document_id=document.id))
+    source = await transcriptions.create_source(
+        Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="attempt-retry.jpg",
+            filename="attempt-retry.jpg",
+            file_path="uploads/attempt-retry.jpg",
+            file_hash="e" * 64,
+            file_size_bytes=1,
+        )
+    )
+    await transcriptions.create_job_source(
+        JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
+    )
+
+    original_flush = AsyncSession.flush
+    execution_attempt_flushes = 0
+
+    async def _flaky_flush(self, *args, **kwargs):
+        nonlocal execution_attempt_flushes
+        if any(isinstance(item, ExecutionAttempt) for item in self.new):
+            execution_attempt_flushes += 1
+            if execution_attempt_flushes == 1:
+                raise IntegrityError("insert execution_attempt", {}, Exception("duplicate attempt number"))
+        return await original_flush(self, *args, **kwargs)
+
+    monkeypatch.setattr(AsyncSession, "flush", _flaky_flush)
+
+    await transcriptions.update_job_source_transcription(
+        job_id=job.id,
+        source_id=source.id,
+        text="retry succeeds",
+        provider="openrouter",
+        model="test-model",
+    )
+
+    assert execution_attempt_flushes == 2
+    rows = await transcriptions.list_job_sources(job_id=job.id)
+    assert len(rows) == 1
+    assert rows[0].status == JobSourceStatus.TRANSCRIBED
+
+
+@pytest.mark.asyncio
+async def test_update_job_source_transcription_raises_domain_error_after_attempt_retry_exhaustion(
+    default_session_factory,
+    monkeypatch,
+):
+    documents = DocumentService(session_factory=default_session_factory)
+    jobs = JobService(session_factory=default_session_factory)
+    transcriptions = SourceService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="attempt-exhaustion-doc"))
+    job = await jobs.create_job(Job(document_id=document.id))
+    source = await transcriptions.create_source(
+        Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="attempt-exhaustion.jpg",
+            filename="attempt-exhaustion.jpg",
+            file_path="uploads/attempt-exhaustion.jpg",
+            file_hash="f" * 64,
+            file_size_bytes=1,
+        )
+    )
+    await transcriptions.create_job_source(
+        JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
+    )
+
+    original_flush = AsyncSession.flush
+
+    async def _always_conflict_flush(self, *args, **kwargs):
+        if any(isinstance(item, ExecutionAttempt) for item in self.new):
+            raise IntegrityError("insert execution_attempt", {}, Exception("duplicate attempt number"))
+        return await original_flush(self, *args, **kwargs)
+
+    monkeypatch.setattr(AsyncSession, "flush", _always_conflict_flush)
+
+    with pytest.raises(TranscriptionError) as exc_info:
+        await transcriptions.update_job_source_transcription(
+            job_id=job.id,
+            source_id=source.id,
+            text="will not persist",
+            provider="openrouter",
+            model="test-model",
+        )
+
+    assert exc_info.value.category == ErrorCategory.CONFLICT
+    assert "attempt number" in exc_info.value.message.lower()

tests/services/test_workflows_reliability.py

@@ -2,6 +2,8 @@
 
 import asyncio
 import time
+from datetime import UTC
+from datetime import datetime
 from pathlib import Path
 from uuid import uuid4
 
@@ -20,6 +22,7 @@ from transcription.db.models import Source
 from transcription.providers.base import TranscriptionResult
 from transcription.services import ServiceBundle
 from transcription.services import workflows as workflows_module
+from transcription.services.workflows import advance_job
 from transcription.services.workflows import process_queued_job
 
 
@@ -366,3 +369,63 @@ class TestWorkflowReliability:
         result = await task
         assert result is not None
         assert result.status == JobStatus.TRANSCRIBED
+
+    @pytest.mark.asyncio
+    async def test_failed_job_with_validation_category_is_not_requeued(self, default_session_factory):
+        services = ServiceBundle.from_session_factory(default_session_factory)
+        async with services.jobs._session_scope() as session:
+            document = Document(id=uuid4(), name="validation-failure-doc")
+            session.add(document)
+            await session.flush()
+
+            source = Source(
+                document_id=document.id,
+                page_number=1,
+                upload_name="validation.jpg",
+                filename="validation.jpg",
+                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
+                file_hash="9" * 64,
+                file_size_bytes=1,
+            )
+            session.add(source)
+            await session.flush()
+
+            job = Job(document_id=document.id, status=JobStatus.FAILED, retry_count=0)
+            session.add(job)
+            await session.flush()
+
+            job_source = JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.FAILED)
+            session.add(job_source)
+            await session.flush()
+
+            now = datetime.now(UTC)
+            session.add(
+                ExecutionAttempt(
+                    job_source_id=job_source.id,
+                    job_id=job.id,
+                    source_id=source.id,
+                    attempt_number=1,
+                    status=JobSourceStatus.FAILED,
+                    provider="fixture",
+                    started_at=now,
+                    finished_at=now,
+                    duration_ms=0,
+                    error_category="validation_error",
+                    error_detail="invalid payload",
+                )
+            )
+            await session.commit()
+            failed_job = await services.jobs.read_job(job_id=job.id, session=session)
+
+        result = await advance_job(
+            failed_job,
+            services=services,
+            settings=Settings(openrouter_api_key="test-key", worker_max_retries=1),
+        )
+
+        assert result is None
+        async with services.jobs._session_scope() as session:
+            persisted = await session.get(Job, failed_job.id)
+            assert persisted is not None
+            assert persisted.status == JobStatus.FAILED
+            assert persisted.retry_count == 0

tests/test_app.py

@@ -28,6 +28,7 @@ class TestAppLifespan:
     def test_startup_initializes_runtime_dependencies(self, monkeypatch, tmp_path):
         """Startup initializes logging, schema, directories, and worker resources."""
         calls = []
+        worker_kwargs = {}
 
         monkeypatch.setattr("transcription.app.configure_logging", lambda _settings: calls.append("logging"))
         monkeypatch.setattr("transcription.app.register_pages", lambda _app: None)
@@ -53,6 +54,7 @@ class TestAppLifespan:
 
         @asynccontextmanager
         async def _worker_lifespan(**_kwargs):
+            worker_kwargs.update(_kwargs)
             calls.append("worker_start")
             yield object(), object(), object()
             calls.append("worker_stop")
@@ -63,6 +65,7 @@ class TestAppLifespan:
             openrouter_api_key="test-key",
             environment="test",
             bootstrap_schema_on_startup=True,
+            worker_poll_interval_seconds=2.5,
             upload_dir=tmp_path / "uploads",
             prompt_dir=tmp_path / "prompts",
         )
@@ -78,6 +81,10 @@ class TestAppLifespan:
         assert "worker_start" in calls
         assert "worker_stop" in calls
         assert "dispose_db" in calls
+        assert worker_kwargs["poll_interval_seconds"] == pytest.approx(settings.worker_poll_interval_seconds)
+        assert worker_kwargs["shutdown_timeout_seconds"] == pytest.approx(
+            settings.worker_provider_timeout_seconds + settings.worker_shutdown_grace_seconds
+        )
         assert settings.upload_dir.exists()
         assert settings.prompt_dir.exists()
 

tests/test_config.py

@@ -167,6 +167,10 @@ class TestWorkerReliabilitySettings:
         """worker retry settings default to no retries."""
         settings = _make_settings()
         assert settings.worker_max_retries == 0
+        assert settings.worker_stale_job_seconds == 30.0
+        assert settings.worker_retry_backoff_seconds == 1.0
+        assert settings.worker_shutdown_grace_seconds == 5.0
+        assert settings.worker_poll_interval_seconds == 1.0
 
 
 def test_provider_timeout_is_not_capped_at_twenty_seconds():

tests/test_meta_contract_guards.py

@@ -167,6 +167,10 @@ def test_env_example_default_values_match_settings_defaults():
         "DATABASE_BACKUP_DIR": str(defaults.database_backup_dir),
         "WORKER_MAX_RETRIES": str(defaults.worker_max_retries),
         "WORKER_PROVIDER_TIMEOUT_SECONDS": str(defaults.worker_provider_timeout_seconds),
+        "WORKER_STALE_JOB_SECONDS": str(defaults.worker_stale_job_seconds),
+        "WORKER_RETRY_BACKOFF_SECONDS": str(defaults.worker_retry_backoff_seconds),
+        "WORKER_SHUTDOWN_GRACE_SECONDS": str(defaults.worker_shutdown_grace_seconds),
+        "WORKER_POLL_INTERVAL_SECONDS": str(defaults.worker_poll_interval_seconds),
         "WORKER_MIN_TRANSCRIPTION_CHARS": str(defaults.worker_min_transcription_chars),
         "WORKER_MIN_TRANSCRIPTION_LINES": str(defaults.worker_min_transcription_lines),
         "WORKER_FAIL_ON_FINISH_REASON_LENGTH": str(defaults.worker_fail_on_finish_reason_length).lower(),

tests/test_worker.py

@@ -1,9 +1,15 @@
 import asyncio
 import logging
+from datetime import UTC
+from datetime import datetime
+from datetime import timedelta
 from typing import cast
 
 import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
 
+from transcription.config import Settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
 from transcription.services import ServiceBundle
@@ -79,6 +85,45 @@ async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog)
     assert worker_health.snapshot().state == "stopped"
 
 
+@pytest.mark.asyncio
+async def test_run_worker_loop_periodically_requeues_stale_processing_jobs(monkeypatch):
+    stop_event = asyncio.Event()
+    stale_sweep_calls: list[datetime] = []
+
+    class _Jobs:
+        settings = Settings(openrouter_api_key="test-key", worker_stale_job_seconds=120.0)
+
+        async def requeue_stale_processing_jobs(self, *, stale_before, session=None):
+            _ = session
+            stale_sweep_calls.append(stale_before)
+            return 1
+
+    class _Bundle:
+        jobs = _Jobs()
+
+        async def aclose(self):
+            return
+
+    monkeypatch.setattr(
+        "transcription.worker.ServiceBundle.from_session_factory",
+        classmethod(lambda _cls, _factory=None, **_kwargs: cast(ServiceBundle, _Bundle())),
+    )
+
+    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
+        _ = (session, session_factory, services)
+        stop_event.set()
+        return False
+
+    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
+
+    fake_session_factory = cast(async_sessionmaker[AsyncSession], object())
+    await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0, session_factory=fake_session_factory)
+
+    assert len(stale_sweep_calls) >= 1
+    expected_upper_bound = datetime.now(UTC) - timedelta(seconds=120.0)
+    assert stale_sweep_calls[0] <= expected_upper_bound
+
+
 @pytest.mark.asyncio
 async def test_run_worker_loop_reuses_one_bundle_across_jobs(monkeypatch):
     """HIGH-02: the provider client is built once per loop, not once per job."""