"""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."