boundaries

.github/instructions/services.instructions.md

@@ -67,6 +67,8 @@ Atomicity rules:
 Separation of concerns:
 
 - Worker modules should stay lightweight and delegate lifecycle transitions to service/workflow orchestration functions.
+- In `workflows.py`, `process_queued_job` should own one complete attempt lifecycle: `QUEUED -> PROCESSING -> TRANSCRIBED|FAILED`.
+- In `workflows.py`, `advance_job` should coordinate broader status progression around attempts (for example retry scheduling from `FAILED -> QUEUED`).
 - Services should expose session-aware write helpers (flush on caller-owned session) so orchestration controls commit boundaries.
 - Backoff/sleep behavior must run outside transactional scopes.
 

src/transcription/models.py

@@ -49,6 +49,11 @@ class Job(SQLModel, table=True):
     document: Document = Relationship(back_populates="jobs")
     transcript: Optional["Transcript"] = Relationship(back_populates="job")
 
+    @property
+    def filename(self) -> str:
+        """Return the filename of the associated document."""
+        return self.document.filename if self.document else "unknown"
+
 
 class Transcript(SQLModel, table=True):
     """The output of a transcription job."""

src/transcription/services/jobs.py

@@ -29,16 +29,17 @@ class JobService(ServiceBase):
     async def read_job(self, job_id: UUID, session: AsyncSession | None = None) -> Job:
         """Read an existing job from the database.
 
-        The selectinload option is used to eagerly load the related document for the job, which makes the full Document
+        The related document is always eagerly loaded so callers can safely
-        model object available in the return Job object.
+        access ``job.document`` in async contexts without triggering lazy-load IO.
         """
         async with self._session_scope(session) as _session:
-            job = await _session.get(
+            query = (
-                Job,
+                select(Job)
-                job_id,
+                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
-                # Makes the full Document model object available in the return Job object
+                .where(Job.id == job_id)
-                options=(selectinload(Job.document),),  # pyright: ignore[reportArgumentType]
+                .execution_options(populate_existing=True)
             )
+            job = (await _session.exec(query)).first()
         if job is None:
             raise ValueError(f"Job with id {job_id} not found")
         return job
@@ -67,7 +68,7 @@ class JobService(ServiceBase):
     ) -> Sequence[Job]:
         """Query jobs from the database based on provided filters."""
         async with self._session_scope(session) as _session:
-            query = select(Job)
+            query = select(Job).options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
             if status is not None:
                 query = query.where(Job.status == status)
             if filename is not None:
@@ -81,11 +82,10 @@ class JobService(ServiceBase):
         load_docs: bool = False,
         session: AsyncSession | None = None,
     ) -> Sequence[Job]:
-        """List all jobs in the database."""
+        """List all jobs in the database with eagerly loaded documents."""
+        _ = load_docs
         async with self._session_scope(session) as _session:
-            query = select(Job)
+            query = select(Job).options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
-            if load_docs:
-                query = query.options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
             result = await _session.exec(query)
             return result.all()
 
@@ -114,7 +114,13 @@ class JobService(ServiceBase):
         once at an orchestration boundary.
         """
         async with self._session_scope(session) as _session:
-            job = await _session.get(Job, job_id)
+            query = (
+                select(Job)
+                .options(selectinload(Job.document))  # pyright: ignore[reportArgumentType]
+                .where(Job.id == job_id)
+                .execution_options(populate_existing=True)
+            )
+            job = (await _session.exec(query)).first()
             if job is None:
                 raise ValueError(f"Job with id {job_id} not found")
             job.status = status

src/transcription/services/workflows.py

@@ -6,7 +6,6 @@ from sqlmodel.ext.asyncio.session import AsyncSession
 from ..config import Settings
 from ..config import get_settings
 from ..errors import AppError
-from ..errors import ErrorCategory
 from ..errors import classify_unexpected_error
 from ..errors import format_error_detail
 from ..models import Job
@@ -23,18 +22,20 @@ async def advance_job(
     job: Job,
     services: ServiceBundle,
     settings: Settings | None = None,
+    session: AsyncSession | None = None,
 ) -> Job | None:
     """Advance a single job by lifecycle status."""
     settings = settings or get_settings()
     match job.status:
         case JobStatus.QUEUED:
-            return await process_job(job=job, services=services, settings=settings)
+            return await process_queued_job(job=job, services=services, session=session)
         case JobStatus.FAILED:
             if job.retry_count < settings.worker_max_retries:
                 return await services.jobs.update_job_state(
                     job_id=job.id,
                     status=JobStatus.QUEUED,
                     retry_count_increment=1,
+                    session=session,
                 )
             else:
                 logger.error(f"Job {job.id} has failed and reached max retries.")
@@ -43,15 +44,13 @@ async def advance_job(
             return
 
 
-async def process_job(
+async def process_queued_job(
     *,
     job: Job,
     services: ServiceBundle,
-    settings: Settings | None = None,
     session: AsyncSession | None = None,
 ) -> Job | None:
-    """Process a queued job with workflow-owned transaction boundaries."""
+    """Process one complete transcription attempt for a queued job."""
-    runtime_settings = settings or get_settings()
     if job.status != JobStatus.QUEUED:
         logger.warning(f"Job {job.id} is not queued. Current status: {job.status}")
         return
@@ -60,28 +59,19 @@ async def process_job(
     if session is None:
         job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING)
     else:
+        # If we're sharing the session, need to make sure setting the Job to PROCESSING is committed before we start
+        # the transcription, otherwise other workers may see the job as still QUEUED and try to process it.
         job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session)
         await session.commit()
 
     document = job.document
-    if document is None:
+    assert document is not None, (
-        error = AppError(
+        f"Job {job.id} has no associated document or the document failed to be loaded by the job service."
-            "Document not found",
+    )
-            category=ErrorCategory.NOT_FOUND,
-            suggestion="Re-upload the source document and retry processing.",
-        )
-        await _finalize_failed(job=job, services=services, error=error, session=session)
-        logger.error(
-            "Job failed operation=worker.process_job job_id=%s error_id=%s category=%s",
-            job.id,
-            error.error_id,
-            error.category.value,
-        )
-        return job
 
     try:
         result = await transcribe_document_image(document.file_path)
-        await _finalize_transcribed(job=job, services=services, result=result, session=session)
+        job = await _finalize_transcribed(job=job, services=services, result=result, session=session)
         logger.info(
             "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s",
             job.id,
@@ -89,27 +79,13 @@ async def process_job(
             result.provider,
         )
     except Exception as exc:  # noqa: BLE001
-        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job")
+        match exc:
-        if _should_retry(job=job, error=error, settings=runtime_settings):
+            case AppError() as error:
-            await _finalize_retry(
+                pass
-                job=job,
+            case _:
-                services=services,
+                error = classify_unexpected_error(exc, operation="worker.process_job")
-                error=error,
-                settings=runtime_settings,
-                session=session,
-            )
-            logger.warning(
-                "Job retried operation=worker.process_job "
-                "job_id=%s document_id=%s retry_count=%s error_id=%s category=%s",
-                job.id,
-                document.id,
-                job.retry_count,
-                error.error_id,
-                error.category.value,
-            )
-            return job
 
-        await _finalize_failed(job=job, services=services, error=error, session=session)
+        job = await _finalize_failed(job=job, services=services, error=error, session=session)
         logger.error(
             "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s",
             job.id,
@@ -131,14 +107,10 @@ async def process_next_queued_job(
     if job is None:
         return False
 
-    await process_job(job=job, services=services, settings=settings, session=session)
+    await advance_job(job=job, services=services, settings=settings, session=session)
     return True
 
 
-def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool:
-    return error.retriable and job.retry_count < settings.worker_max_retries
-
-
 async def _finalize_transcribed(
     *,
     job: Job,

src/transcription/worker.py

@@ -21,7 +21,7 @@ from .services import ServiceBundle
 from .services.documents import DocumentService
 from .services.jobs import JobService
 from .services.transcription import TranscriptionService
-from .services.workflows import process_job
+from .services.workflows import advance_job
 from .services.workflows import process_next_queued_job as process_next_queued_job_workflow
 
 logger = logging.getLogger(__name__)
@@ -37,7 +37,7 @@ async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Ev
         with handle_worker_exceptions():
             async with _get_queue_item(queue) as job_id:
                 job = await service.read_job(job_id)
-                asyncio.create_task(process_job(job=job, services=ServiceBundle()))
+                asyncio.create_task(advance_job(job=job, services=ServiceBundle()))
 
 
 @contextmanager