transcription/tests/integration/test_pipeline_atomicity.py

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