feat: run stale-job recovery periodically

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

.env.example

@@ -57,6 +57,7 @@ DATABASE_BACKUP_DIR=./data/backups
 # --- worker reliability ---
 WORKER_MAX_RETRIES=0
 WORKER_PROVIDER_TIMEOUT_SECONDS=30.0
+WORKER_STALE_JOB_SECONDS=30.0
 WORKER_MIN_TRANSCRIPTION_CHARS=0
 WORKER_MIN_TRANSCRIPTION_LINES=0
 WORKER_FAIL_ON_FINISH_REASON_LENGTH=false

src/transcription/app.py

@@ -71,11 +71,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,7 @@ 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_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/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
@@ -215,6 +218,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/test_config.py

@@ -167,6 +167,7 @@ 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
 
 
 def test_provider_timeout_is_not_capped_at_twenty_seconds():

tests/test_meta_contract_guards.py

@@ -167,6 +167,7 @@ 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_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."""