"""Reliability tests for worker workflow timeout behavior."""

import asyncio
import time
from pathlib import Path
from uuid import uuid4

import pytest
from sqlmodel import col
from sqlmodel import select

from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import ExecutionAttempt
from transcription.db.models import Job
from transcription.db.models import JobSource
from transcription.db.models import JobSourceStatus
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 import workflows as workflows_module
from transcription.services.workflows import process_queued_job


@pytest.mark.integration
class TestWorkflowReliability:
    """Verify timeout and terminal-state reliability behavior."""

    @pytest.mark.asyncio
    async def test_process_queued_job_timeout_marks_job_failed(self, default_session_factory, monkeypatch):
        """Provider timeout transitions a queued job to failed with error detail."""
        services = ServiceBundle()
        object.__setattr__(
            services,
            "documents",
            services.documents.__class__(session_factory=default_session_factory),
        )
        object.__setattr__(
            services,
            "jobs",
            services.jobs.__class__(session_factory=default_session_factory),
        )
        object.__setattr__(
            services,
            "sources",
            services.sources.__class__(session_factory=default_session_factory),
        )

        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="timeout-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="timeout.jpg",
                filename="timeout.jpg",
                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                file_hash="c" * 64,
                file_size_bytes=1,
            )
            session.add(source)
            await session.flush()
            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
            await session.commit()

            loaded = await services.jobs.read_job(job_id=job.id, session=session)

        async def _never_returns(
            image_path,
            *,
            prompt_name="transcribe_document.md",
            prompt_text=None,
            temperature=None,
            top_p=None,
            settings=None,
            provider=None,
        ):
            _ = (image_path, prompt_name, prompt_text, temperature, top_p, settings, provider)
            raise TimeoutError("simulated provider timeout")

        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns)

        timeout_settings = Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=20.0)
        result = await process_queued_job(job=loaded, services=services, settings=timeout_settings)

        assert result is not None
        assert result.status == JobStatus.FAILED

        async with services.jobs._session_scope() as session:
            attempts = (
                await session.execute(
                    select(ExecutionAttempt).where(
                        col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                    )
                )
            ).scalars().all()
        error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None)
        assert "timed out" in error_detail.lower()
        assert "20.0s" in error_detail

    @pytest.mark.asyncio
    async def test_timeout_duration_excludes_pre_call_setup(self, default_session_factory, monkeypatch):
        """duration_ms covers only the provider call, not the setup preceding it.

        Regression guard for review log [55]: three historical ``local_timeout`` rows
        recorded 0.4-2.0 s more than the configured budget because the measurement
        window opened before payload resolution. Blocking setup is simulated here so
        the assertion fails if that window ever reopens.
        """
        services = ServiceBundle.from_session_factory(default_session_factory)
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="window-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="window.jpg",
                filename="window.jpg",
                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                file_hash="d" * 64,
                file_size_bytes=1,
            )
            session.add(source)
            await session.flush()
            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
            await session.commit()
            loaded = await services.jobs.read_job(job_id=job.id, session=session)

        setup_seconds = 0.40
        budget_seconds = 0.20
        real_build = workflows_module.build_provider_input

        def _slow_build(source_arg):
            time.sleep(setup_seconds)
            return real_build(source_arg)

        async def _never_returns(*args, **kwargs):
            _ = (args, kwargs)
            await asyncio.sleep(budget_seconds * 20)

        monkeypatch.setattr("transcription.services.workflows.build_provider_input", _slow_build)
        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns)

        result = await process_queued_job(
            job=loaded,
            services=services,
            settings=Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=budget_seconds),
        )
        assert result is not None
        assert result.status == JobStatus.FAILED

        async with services.jobs._session_scope() as session:
            attempts = (
                (
                    await session.exec(
                        select(ExecutionAttempt).where(
                            col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
                        )
                    )
                )
                .all()
            )
        assert len(attempts) == 1
        duration_ms = attempts[0].duration_ms

        # At or just above the budget, and well clear of budget + setup.
        assert duration_ms >= int(budget_seconds * 1000 * 0.9)
        assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9)

    @pytest.mark.asyncio
    async def test_error_after_claim_fails_the_job_instead_of_stranding_it(
        self,
        default_session_factory,
        monkeypatch,
    ):
        """A non-retriable fault after the claim drives the job terminal, not stuck.

        Regression guard for review log [8]. The claim commits PROCESSING before any
        provider work, and claim_next_queued_job only ever selects QUEUED, so an
        exception escaping advance_job used to strand the job in PROCESSING forever
        with one swallowed log line. Measured before the fix: raised once, job left
        processing, retry_count 0, never re-claimed.
        """
        services = ServiceBundle.from_session_factory(default_session_factory)
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="strand-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="strand.jpg",
                filename="strand.jpg",
                file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                file_hash="e" * 64,
                file_size_bytes=1,
            )
            session.add(source)
            await session.flush()
            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
            await session.commit()
            job_id = job.id

        async def _succeeds(*args, **kwargs):
            _ = (args, kwargs)
            return TranscriptionResult(text="page text", provider="test", model="test-model")

        async def _boom(**kwargs):
            _ = kwargs
            raise AttributeError("deliberate programming error")

        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds)
        monkeypatch.setattr("transcription.services.workflows._finalize_batch_outcome", _boom)

        processed = await workflows_module.process_next_queued_job(services=services)

        assert processed is True
        async with services.jobs._session_scope() as session:
            final = await session.get(Job, job_id)
            assert final is not None
            # Terminal and resubmittable, rather than stranded in PROCESSING.
            assert final.status == JobStatus.FAILED

    @pytest.mark.asyncio
    async def test_completed_page_is_committed_before_next_provider_call_finishes(
        self,
        default_session_factory,
        monkeypatch,
    ):
        services = ServiceBundle.from_session_factory(default_session_factory)
        async with services.jobs._session_scope() as session:
            document = Document(id=uuid4(), name="durability-doc")
            session.add(document)
            await session.flush()
            job = Job(document_id=document.id, status=JobStatus.QUEUED)
            session.add(job)
            await session.flush()
            for page_number in (1, 2):
                source = Source(
                    document_id=document.id,
                    page_number=page_number,
                    upload_name=f"page-{page_number}.jpg",
                    filename=f"page-{page_number}.jpg",
                    file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
                    file_hash=str(page_number) * 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()
            loaded = await services.jobs.read_job(job_id=job.id, session=session)

        second_started = asyncio.Event()
        release_second = asyncio.Event()
        call_count = 0

        async def _transcribe(image_path, **kwargs):
            nonlocal call_count
            _ = (image_path, kwargs)
            call_count += 1
            if call_count == 2:
                second_started.set()
                await release_second.wait()
            return TranscriptionResult(text=f"page {call_count}", provider="fixture", model="model")

        monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _transcribe)
        task = asyncio.create_task(process_queued_job(job=loaded, services=services))
        await asyncio.wait_for(second_started.wait(), timeout=2)

        attempts = await services.evidence.list_execution_attempts(job_id=job.id)
        assert len(attempts) == 1
        assert attempts[0].raw_transcription == "page 1"

        release_second.set()
        result = await task
        assert result is not None
        assert result.status == JobStatus.TRANSCRIBED