apps/api/tests/test_reliability.py

"""Failure classification and the honesty of the ledger under a lost race.

Two regressions that only show up when something goes wrong: a stalled model call
was classified two different ways depending on which of two identical deadlines
fired first, and a status transition that lost a race still wrote itself into the
append-only audit log as though it had happened.
"""

from __future__ import annotations

import asyncio
import uuid
from decimal import Decimal

import pytest
from sqlalchemy import func, select

from app.core.claude import LiveClaudeClient
from app.core.db import transaction
from app.core.retry import RetriesExhaustedError, RetryPolicy, run_with_retry
from app.core.settings import Settings
from app.models.enums import AuditEvent, DocumentStatus
from app.models.tables import AuditLog, Document
from app.pipeline.orchestrator import PipelineOrchestrator

# --- F-2: a wait_for timeout is retryable, and the SDK trips first ----------


def test_outer_bound_sits_above_the_sdk_timeout_so_they_cannot_race() -> None:
    settings = Settings(anthropic_api_key="sk-ant-not-a-real-key", llm_mode="live")
    client = LiveClaudeClient(settings)
    assert client._policy.timeout_s > settings.llm_timeout_s
    assert TimeoutError in client._policy.retry_on


async def test_wait_for_timeout_is_retried() -> None:
    calls = 0

    async def stalls() -> None:
        nonlocal calls
        calls += 1
        await asyncio.sleep(1.0)

    policy = RetryPolicy(attempts=2, base_delay_s=0.0, timeout_s=0.01, retry_on=(TimeoutError,))
    with pytest.raises(RetriesExhaustedError):
        await run_with_retry("stalls", stalls, policy)

    assert calls == 2, "the outer timeout must consume the retry budget, not bypass it"


async def test_without_timeout_error_the_budget_is_bypassed() -> None:
    """Documents the defect: `wait_for` raises the built-in `TimeoutError`, which is
    unrelated to `anthropic.APITimeoutError`, so a policy listing only the SDK's
    error abandons the call on attempt one and reports it as an internal fault."""
    calls = 0

    async def stalls() -> None:
        nonlocal calls
        calls += 1
        await asyncio.sleep(1.0)

    policy = RetryPolicy(attempts=3, base_delay_s=0.0, timeout_s=0.01, retry_on=(ConnectionError,))
    with pytest.raises(TimeoutError):
        await run_with_retry("stalls", stalls, policy)

    assert calls == 1


# --- F-3: a lost transition is not written into the audit log --------------


@pytest.mark.integration
async def test_only_the_winning_transition_is_audited(
    orchestrator: PipelineOrchestrator, clean_db: None
) -> None:
    """Two finalisers race one PROCESSING row; exactly one transition is real.

    Both read `PROCESSING`, both pass the transition check, and both attempt the
    conditional UPDATE. Postgres lets one through and the loser matches zero rows —
    at which point it must not claim a state change, because the append-only trigger
    makes anything it writes permanent.
    """
    document_id = uuid.uuid4()
    async with transaction() as session:
        session.add(
            Document(
                id=document_id,
                file_hash="f" * 64,
                filename="race.pdf",
                media_type="application/pdf",
                size_bytes=1024,
                status=str(DocumentStatus.PROCESSING),
            )
        )

    await asyncio.gather(
        orchestrator._finalise(
            document_id=document_id,
            target=DocumentStatus.DONE,
            reason=None,
            latency_ms=10,
            cost_usd=Decimal("0"),
        ),
        orchestrator._finalise(
            document_id=document_id,
            target=DocumentStatus.NEEDS_REVIEW,
            reason="review",
            latency_ms=10,
            cost_usd=Decimal("0"),
        ),
    )

    async with transaction() as session:
        transitions = (
            await session.execute(
                select(func.count())
                .select_from(AuditLog)
                .where(
                    AuditLog.document_id == document_id,
                    AuditLog.event == str(AuditEvent.STATUS_CHANGED),
                )
            )
        ).scalar_one()
        final = (
            await session.execute(select(Document.status).where(Document.id == document_id))
        ).scalar_one()

    assert transitions == 1, "the losing finaliser must not audit a transition it never made"
    assert final in {str(DocumentStatus.DONE), str(DocumentStatus.NEEDS_REVIEW)}