arm-core/api/tests/test_result_mapping.py

"""Цикл маппинга результата распознавания в `document`/`osv_entry` (spec osv-result-mapping; design D-1, D-2, D-3, D-7,
D-9; рулинги R-3, R-4, R-5; Review Focus 1, 2).

Интеграционные тесты — на отдельной временной БД (`committed_sessionmaker`: цикл работает в собственных транзакциях),
результат лежит в `FakeStorage` под `results/{file_id}/{job_id}/result.json`. Синтетические результаты строятся из
примера контракта `contracts/examples/v1/result-osv-minimal.json`; сценарий «Результат смапплен» и замер скорости — на
golden-снимках парсера (маркер `fixtures`).
"""

import asyncio
import copy
import json
import time
import uuid
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any

import pytest
from loguru import logger
from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from src.contract import ack_msg_id, result_msg_id
from src.contract.result import RecognitionResult
from src.core.config import Settings, get_settings
from src.domain.consolidation_periods import reporting_date
from src.models.access import AppUser
from src.models.accounting import ChartOfAccount
from src.models.audit_ex import AuditEvent
from src.models.consolidation import (
    Consolidation,
    ConsolidationDocument,
    ConsolidationStatusHistory,
    ConsolidationVersion,
)
from src.models.directories import LegalEntity
from src.models.enums import (
    AuditAction,
    AuditSection,
    ConsolidationStatus,
    CounterpartyMatchKind,
    DocumentCheckResult,
    DocumentStatus,
    DocumentType,
    FileStatus,
    OsvDataType,
    OsvGroupType,
    PeriodQuarter,
    RecognitionStatus,
)
from src.models.files import Document, RecognitionJob, StoredFile
from src.models.osv import OsvEntry
from src.models.statements import (
    DocumentIssue,
    FinancialStatementLine,
    FinancialStatementMeta,
    FinancialStatementValue,
)
from src.repositories.counterparties import CounterpartyIndex
from src.services import consolidation_documents
from src.services import result_mapping as rm
from src.services.recognition_exchange import expire_deadlines, handle_message
from src.services.result_mapping import (
    MAPPING_ATTEMPTS_KEY,
    MAPPING_ERROR_KEY,
    MAPPING_WARNINGS_KEY,
    NO_DOCUMENTS,
    Candidate,
    MappingFatal,
    process_once,
    run_mapping_loop,
    supersede_previous,
    validate_result_key,
)
from tests.conftest import FakeStorage
from tests.factories import (
    make_company_group,
    make_consolidation,
    make_document,
    make_legal_entity,
    make_result_object,
    make_stored_file,
    make_version,
)
from tests.test_recognition_inbox import envelope, fake_msg

API_DIR = Path(__file__).resolve().parents[1]
MINIMAL_RESULT = API_DIR / "contracts" / "examples" / "v1" / "result-osv-minimal.json"
GOLDEN_OSV_66 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_66 за 2024.json"
GOLDEN_OSV_60 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_60 за 2024.json"
BUCKET = "arm-ka"
OWNER_NAME = "АО «Русский бисквит»"
FACTORY_HISTORY = 2  # `make_consolidation(status=DOCUMENTS_PROCESSING)`: NULL → INITIATED → DOCUMENTS_PROCESSING
PERF_BUDGET_SECONDS = 2.0  # цель плана: маппинг ~1000 строк ОСВ_60 за 2024 быстрее 2 с

pytestmark = pytest.mark.integration


# --- данные ---------------------------------------------------------------------------------------------------------


def _minimal() -> dict[str, Any]:
    return json.loads(MINIMAL_RESULT.read_text("utf-8"))


def _golden(fixtures_dir: Path, path: Path) -> dict[str, Any]:
    data = json.loads((fixtures_dir / path).read_text("utf-8"))
    # у снимка нет job_id/file_id (подставляются) и версии сервиса (в контракте она обязательна)
    data["service"] = {"version": "golden", **data["service"]}
    return data


def _osv(result: dict[str, Any]) -> dict[str, Any]:
    return result["documents"][0]


def _root(rows: list[dict[str, Any]]) -> dict[str, Any]:
    return next(row for row in rows if row["level"] == 0 and row["parent_row_number"] is None)


def _settings() -> Settings:
    return get_settings()


def _now() -> datetime:
    return datetime.now(UTC)


async def _prepare(  # noqa: PLR0913 - каждое поле подготовки задаётся тестом явно
    maker: async_sessionmaker[AsyncSession],
    storage: FakeStorage,
    result: dict[str, Any],
    *,
    entity: LegalEntity | None = None,
    owner_name: str = OWNER_NAME,
    owner_inn: str | None = None,
    job_status: RecognitionStatus = RecognitionStatus.done,
    file_status: FileStatus = FileStatus.PROCESSING,
    key: str | None = None,
    uploaded_by: uuid.UUID | None = None,
) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
    """ЮЛ-владелец, файл `PROCESSING` с текущим заданием `done`, результат в `storage`; `(file_id, job_id, entity_id)`.

    `job_id`/`file_id` результата подставляются; `key` — подменить ключ `s3_object` результата (OQ-70)."""
    async with maker() as session, session.begin():
        owner = entity or await make_legal_entity(session, name=owner_name, inn=owner_inn)
        file, job = await make_stored_file(
            session, entity=owner, status=file_status, job_status=job_status, uploaded_by=uploaded_by
        )
        job.finished_at = _now()
        job.metrics = copy.deepcopy(result.get("metrics"))  # как inbox: метрики сервиса из конверта
        payload = copy.deepcopy(result) | {"job_id": str(job.id), "file_id": str(file.id)}
        s3 = await make_result_object(session, job=job, storage=storage, result=payload)
        if key is not None:
            storage.objects[key] = storage.objects[s3.key]
            s3.key = key
        return file.id, job.id, owner.id


async def _file(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> StoredFile:
    async with maker() as session:
        file = await session.get(StoredFile, file_id)
        assert file is not None
        return file


async def _job(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> RecognitionJob:
    async with maker() as session:
        job = await session.get(RecognitionJob, job_id)
        assert job is not None
        return job


async def _documents(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> list[Document]:
    async with maker() as session:
        rows = await session.execute(select(Document).where(Document.stored_file_id == file_id))
        return list(rows.scalars())


async def _entries(maker: async_sessionmaker[AsyncSession], document_id: uuid.UUID) -> list[OsvEntry]:
    async with maker() as session:
        rows = await session.execute(
            select(OsvEntry).where(OsvEntry.document_id == document_id).order_by(OsvEntry.row_number)
        )
        return list(rows.scalars())


# --- проверка ключа результата (без БД) ----------------------------------------------------------------------------


def _candidate(key: str, bucket: str = BUCKET) -> Candidate:
    return Candidate(job_id=JOB_ID, file_id=FILE_ID, bucket=bucket, key=key)


JOB_ID = uuid.UUID("3f2a9c1e-7b4d-4e0a-9d51-2c8e6a1b0f01")
FILE_ID = uuid.UUID("a1c4e7d2-5b36-4f89-8e12-9d0c3b7a4e02")


@pytest.mark.parametrize(
    "key",
    [f"results/{FILE_ID}/{JOB_ID}/result.json", f"results/{FILE_ID}/{JOB_ID}/sub/result.json"],
)
def test_validate_result_key_accepts_job_folder(key: str) -> None:
    validate_result_key(_candidate(key), BUCKET)


@pytest.mark.parametrize(
    ("key", "bucket"),
    [
        (f"results/{uuid.uuid4()}/{JOB_ID}/result.json", BUCKET),  # чужой файл
        (f"results/{FILE_ID}/{uuid.uuid4()}/result.json", BUCKET),  # чужое задание
        (f"uploads/{FILE_ID}/{JOB_ID}/result.json", BUCKET),  # не папка результатов
        (f"results/{FILE_ID}/{JOB_ID}/", BUCKET),  # нет имени объекта
        (f"results/{FILE_ID}/{JOB_ID}/../{uuid.uuid4()}/result.json", BUCKET),  # выход из папки
        (f"results/{FILE_ID}/{JOB_ID}/result.json", "other-bucket"),  # чужой бакет
    ],
)
def test_validate_result_key_rejects_outside_job_folder(key: str, bucket: str) -> None:
    with pytest.raises(MappingFatal):
        validate_result_key(_candidate(key, bucket), BUCKET)


# --- сценарии спеки ------------------------------------------------------------------------------------------------


@pytest.mark.fixtures
async def test_golden_osv_66_mapped(  # noqa: PLR0915 - один сценарий спеки целиком
    committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path
):
    """«Результат смапплен», «Дерево и счета»: документ 66/2024/Q4/нарастающая, строки без TOTAL, `group_schema`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    golden = _golden(fixtures_dir, GOLDEN_OSV_66)
    file_id, job_id, entity_id = await _prepare(maker, storage, golden)

    assert await process_once(maker, storage, _settings()) is True

    file = await _file(maker, file_id)
    assert file.status is FileStatus.PROCESSED
    assert file.updated_at is not None
    [document] = await _documents(maker, file_id)
    assert document.document_type is DocumentType.OSV
    assert document.osv_account_code == "66"
    assert document.period_year == 2024  # noqa: PLR2004 - год из шапки снимка
    assert document.period_quarter is PeriodQuarter.Q4
    assert document.is_cumulative is True
    assert document.locator == "sheet:Лист_1"
    assert document.legal_entity_id == entity_id
    assert document.status is DocumentStatus.CONFIRMED  # done, чистые проверки, шапка = владелец → автоподтверждение
    assert document.counterparty_check is DocumentCheckResult.MATCH
    assert document.period_check is DocumentCheckResult.UNKNOWN

    rows = _osv(golden)["rows"]
    entries = await _entries(maker, document.id)
    assert len(entries) == len(rows) - 1  # всё, кроме TOTAL
    total_rows = {row["row_number"] for row in rows if row["group_type"] == "TOTAL"}
    assert total_rows and not total_rows & {entry.row_number for entry in entries}  # в enum DBML TOTAL нет
    by_row = {entry.row_number: entry for entry in entries}
    for row in rows:
        if row["group_type"] == "TOTAL":
            continue
        entry = by_row[row["row_number"]]
        assert entry.recognition_job_id == job_id
        assert entry.owner_legal_entity_id == entity_id
        assert entry.is_modified is False
        assert entry.level == row["level"]
        assert entry.row_value == row["row_value"]
        for name in ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit"):
            expected = row.get(name)
            assert getattr(entry, name) == (Decimal(expected) if expected is not None else None)
        if row["parent_row_number"] is None:
            assert entry.parent_entry_id is None
        else:
            assert entry.parent_entry_id == by_row[row["parent_row_number"]].id

    async with maker() as session:
        chart_6603 = (await session.execute(select(ChartOfAccount.id).where(ChartOfAccount.code == "66.03"))).scalar()
    [yug] = [e for e in entries if e.account_code == "66.03" and e.group_type is OsvGroupType.COUNTERPARTY]
    assert yug.counterparty_name == "Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ"
    parent = next(e for e in entries if e.id == yug.parent_entry_id)
    assert parent.account_code == "66.03"
    assert parent.group_type is OsvGroupType.ACCOUNT
    assert yug.account_root == "66"
    assert yug.account_id == chart_6603 is not None

    job = await _job(maker, job_id)
    assert job.group_schema == [
        {
            "locator": "sheet:Лист_1",
            "groups": [
                {"type": "ACCOUNT", "name": "Счет", "has_inn": None},
                {"type": "COUNTERPARTY", "name": "Контрагенты", "has_inn": None},
            ],
        }
    ]
    assert job.metrics[MAPPING_WARNINGS_KEY] == []
    assert job.metrics["rows"] == 9  # noqa: PLR2004 - метрики сервиса сохранены рядом с ключами маппинга


@pytest.mark.fixtures
async def test_mapping_performance_osv_60(committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path):
    """Цель плана: маппинг самой длинной ОСВ (`ОСВ_60 за 2024`, ~1000 строк) — быстрее 2 с на один проход цикла."""
    maker, storage = committed_sessionmaker, FakeStorage()
    golden = _golden(fixtures_dir, GOLDEN_OSV_60)
    file_id, _, _ = await _prepare(maker, storage, golden)

    started = time.perf_counter()
    assert await process_once(maker, storage, _settings()) is True
    elapsed = time.perf_counter() - started

    [document] = await _documents(maker, file_id)
    rows = _osv(golden)["rows"]
    entries = await _entries(maker, document.id)
    total_rows = sum(1 for row in rows if row["group_type"] == "TOTAL")
    assert len(entries) == len(rows) - total_rows
    logger.info("Маппинг ОСВ_60 за 2024: {rows} строк за {elapsed:.3f} с", rows=len(entries), elapsed=elapsed)
    print(f"\nPERF: ОСВ_60 за 2024 — {len(entries)} строк за {elapsed:.3f} с")  # noqa: T201 - замер в отчёт
    assert elapsed < PERF_BUDGET_SECONDS


async def test_result_key_outside_job_folder_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """«Ключ результата вне папки задания»: файл `ERROR` с ошибкой маппинга, документов нет, повтора нет."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(
        maker, storage, _minimal(), key=f"results/{uuid.uuid4()}/{uuid.uuid4()}/result.json"
    )

    assert await process_once(maker, storage, _settings()) is True

    file = await _file(maker, file_id)
    assert file.status is FileStatus.ERROR
    job = await _job(maker, job_id)
    assert "results/" in job.metrics[MAPPING_ERROR_KEY]
    assert MAPPING_ATTEMPTS_KEY not in job.metrics  # неповторяемая ошибка — без счётчика попыток
    assert await _documents(maker, file_id) == []
    assert await process_once(maker, storage, _settings()) is False  # повтора нет


async def test_missing_result_object_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Объекта результата нет в S3 — неповторяемая ошибка (D-2): файл `ERROR`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
    storage.objects.clear()

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    assert "не найден" in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]


async def test_result_with_empty_documents_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Результат без документов не проходит схему контракта (documents ≥ 1) → файл `ERROR`, не исключение цикла."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal() | {"status": "needs_check", "documents": []}
    file_id, job_id, _ = await _prepare(maker, storage, result, job_status=RecognitionStatus.needs_check)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    assert "схем" in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]
    assert await _documents(maker, file_id) == []


async def test_map_file_without_documents_is_fatal(seeded: AsyncSession):
    """`map_file` на результате без документов (схема такого не пропускает — защита) → `NO_DOCUMENTS`."""
    entity = await make_legal_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.done
    )
    result = RecognitionResult.model_construct(documents=[])
    index = await CounterpartyIndex.load(seeded)
    with pytest.raises(MappingFatal, match=NO_DOCUMENTS):
        await rm.map_file(seeded, job=job, file=file, result=result, chart={}, index=index, now=datetime.now(UTC))


async def test_mapping_failure_retries_then_error(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """«Сбой маппинга с повтором»: откат (документов нет), `mapping_attempts` 1, 2 — файл `PROCESSING`; третья
    неудача — `ERROR` с `metrics.mapping_error`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
    real_map_file = rm.map_file
    calls: list[int] = []

    async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
        await real_map_file(session, **kwargs)  # вставка выполнена — и откатывается вместе с транзакцией
        calls.append(1)
        raise RuntimeError("подставная ошибка БД при вставке строк")

    monkeypatch.setattr(rm, "map_file", failing_map_file)

    for attempt in (1, 2):
        # R-15: повторяемая неудача — `False`, цикл выдерживает паузу MAPPING_POLL_SECONDS до следующей попытки
        assert await process_once(maker, storage, _settings()) is False
        job = await _job(maker, job_id)
        assert job.metrics[MAPPING_ATTEMPTS_KEY] == attempt
        assert MAPPING_ERROR_KEY not in job.metrics
        assert (await _file(maker, file_id)).status is FileStatus.PROCESSING
        assert await _documents(maker, file_id) == []

    # третья неудача переводит файл в ERROR: кандидат покинул очередь, следующий можно брать без паузы — `True`
    assert await process_once(maker, storage, _settings()) is True
    job = await _job(maker, job_id)
    assert job.metrics[MAPPING_ATTEMPTS_KEY] == 3  # noqa: PLR2004 - MAPPING_MAX_ATTEMPTS по умолчанию
    # M-1: пользователю — категория и число попыток, текст исключения остаётся только в логе
    assert job.metrics[MAPPING_ERROR_KEY] == "маппинг не удался: ошибка БД/S3 (RuntimeError), попыток 3"
    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    assert len(calls) == 3  # noqa: PLR2004 - три попытки
    assert await process_once(maker, storage, _settings()) is False


_INTERNALS = ("[SQL:", "input_value", "http", "sqlalche.me", "parameters")


async def test_schema_error_hides_input_values(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """M-1: ошибка схемы результата — в `mapping_error` число ошибок и путь/тип первой, без `input_value` и ссылок
    pydantic (полный текст — только в логе)."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal()
    _osv(result)["rows"][1]["opening_debit"] = "секрет http://example.invalid"
    file_id, job_id, _ = await _prepare(maker, storage, result)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]
    assert message.startswith("результат не соответствует схеме recognition-result: ошибок ")
    assert "первая: documents.0." in message
    assert "opening_debit" in message
    assert "секрет" not in message
    assert not [marker for marker in _INTERNALS if marker in message]


async def test_retryable_db_error_hides_sql(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """M-1: исчерпавшая попытки ошибка БД — в `mapping_error` категория, тип исключения и число попыток, без
    `[SQL: …]`, параметров и ссылок SQLAlchemy."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())

    async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
        raise IntegrityError("INSERT INTO osv_entry (id) VALUES ($1)", {"id": "секрет"}, Exception("fk violation"))

    monkeypatch.setattr(rm, "map_file", failing_map_file)
    settings = _settings().model_copy(update={"MAPPING_MAX_ATTEMPTS": 1})

    assert await process_once(maker, storage, settings) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]
    assert message == "маппинг не удался: ошибка БД/S3 (IntegrityError), попыток 1"
    assert not [marker for marker in _INTERNALS if marker in message]


async def test_child_before_parent_across_batches(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """T7.2: строки вставляются родитель раньше потомка независимо от порядка строк в результате — потомок перед
    родителем на границе пачки не нарушает self-FK `parent_entry_id`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal()
    rows = _osv(result)["rows"]
    rows.reverse()  # потомок (level 1) раньше корня (level 0)
    monkeypatch.setattr(rm, "ENTRY_BATCH_SIZE", 1)  # каждая строка — отдельная пачка
    file_id, _, _ = await _prepare(maker, storage, result)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
    [document] = await _documents(maker, file_id)
    entries = await _entries(maker, document.id)
    root, child = entries
    assert child.parent_entry_id == root.id


async def test_repeated_mapping_is_idempotent(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла
    ровно один набор документов и строк."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal()
    rows = _osv(result)["rows"]
    rows.extend(
        rows[-1] | {"row_number": number, "counterparty_name": f"ООО Лютик-{number}", "counterparty_inn": None}
        for number in range(3, 7)
    )
    file_id, job_id, entity_id = await _prepare(maker, storage, result)
    async with maker() as session, session.begin():
        file = await session.get(StoredFile, file_id)
        job = await session.get(RecognitionJob, job_id)
        assert file is not None and job is not None
        stale = await make_document(session, file=file, job=job, account="60", status=DocumentStatus.NEEDS_CHECK)
        for row in rows[: len(rows) // 2]:
            session.add(
                OsvEntry(
                    recognition_job_id=job_id,
                    document_id=stale.id,
                    owner_legal_entity_id=entity_id,
                    row_number=row["row_number"],
                    level=row["level"],
                    row_value=row["row_value"],
                )
            )

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.id != stale.id
    entries = await _entries(maker, document.id)
    assert [entry.row_number for entry in entries] == [row["row_number"] for row in rows]
    async with maker() as session:
        total = (
            await session.execute(
                select(func.count())
                .select_from(OsvEntry)
                .join(Document, Document.id == OsvEntry.document_id)
                .where(Document.stored_file_id == file_id)
            )
        ).scalar_one()
    assert total == len(rows)


async def test_two_sheets_of_same_account(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """«Два листа одного счёта»: один документ (первый лист), `NEEDS_CHECK`, предупреждение называет оба `locator`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal()
    first = _osv(result)
    second = copy.deepcopy(first) | {"locator": "sheet:Лист_2"}
    second["header"]["account_code_raw"] = "60"
    result["documents"] = [first, second]
    file_id, job_id, _ = await _prepare(maker, storage, result, owner_inn="7700000000")

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.locator == "sheet:Лист_1"
    assert document.osv_account_code == "60"
    assert document.status is DocumentStatus.NEEDS_CHECK
    assert len(await _entries(maker, document.id)) == len(first["rows"])
    job = await _job(maker, job_id)
    assert job.metrics[MAPPING_WARNINGS_KEY] == ["sheet=sheet:Лист_2: счёт 60 уже есть в листе sheet:Лист_1"]
    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED


async def test_header_of_another_organization(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """«Шапка другой организации»: файл ЮЛ «Пекарня Счастливочная», в шапке `АО "Русский бисквит"` → `MISMATCH`,
    `NEEDS_CHECK`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    result = _minimal()
    _osv(result)["header"] |= {"organization_name": 'АО "Русский бисквит"', "organization_inn": None}
    file_id, _, _ = await _prepare(maker, storage, result, owner_name="ООО «Пекарня Счастливочная»")

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.counterparty_check is DocumentCheckResult.MISMATCH
    assert document.status is DocumentStatus.NEEDS_CHECK


async def _q1_document(
    maker: async_sessionmaker[AsyncSession],
    entity_id: uuid.UUID,
    *,
    opening_credit: Decimal,
    file_status: FileStatus,
) -> None:
    """Документ ОСВ 60 за 2024 Q1 другого файла того же ЮЛ с корнем (level 0) и строкой-потомком."""
    async with maker() as session, session.begin():
        entity = await session.get(LegalEntity, entity_id)
        assert entity is not None
        file, job = await make_stored_file(
            session, entity=entity, status=file_status, job_status=RecognitionStatus.done
        )
        document = await make_document(session, file=file, job=job, account="60", year=2024, quarter=PeriodQuarter.Q1)
        common = {"recognition_job_id": job.id, "document_id": document.id, "owner_legal_entity_id": entity_id}
        root = OsvEntry(
            **common, row_number=5, level=0, row_value="60", opening_debit=None, opening_credit=opening_credit
        )
        session.add(root)
        await session.flush()
        # строка уровня 1 с меньшим номером не должна считаться корнем
        session.add(OsvEntry(**common, row_number=1, level=1, parent_entry_id=root.id, opening_credit=Decimal(1)))


@pytest.mark.parametrize(
    ("q1_opening", "q1_file_status", "expected"),
    [
        (Decimal("1000.00"), FileStatus.PROCESSED, True),  # сальдо на начало совпало с Q1
        (Decimal("1000.01"), FileStatus.PROCESSED, True),  # в пределах ±0.01
        (Decimal("999.00"), FileStatus.PROCESSED, False),  # не совпало — по тексту
        (Decimal("1000.00"), FileStatus.SUPERSEDED, False),  # Q1 из заменённого файла не учитывается
    ],
)
async def test_cumulative_by_q1_data(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
    q1_opening: Decimal,
    q1_file_status: FileStatus,
    expected: bool,
):
    """«Нарастающая по данным»: ОСВ 60 «за 3 квартал 2024» с `opening_*` корня = Q1 → `is_cumulative = true` и
    предупреждение «нарастающая по данным Q1» (ASSUMPTION(OQ-02))."""
    maker, storage = committed_sessionmaker, FakeStorage()
    async with maker() as session, session.begin():
        entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000")
    await _q1_document(maker, entity.id, opening_credit=q1_opening, file_status=q1_file_status)
    result = _minimal()
    _osv(result)["header"] |= {
        "period_text": "3 квартал 2024 г.",
        "period_year": 2024,
        "period_quarter": "Q3",
        "is_cumulative": False,
    }
    assert _root(_osv(result)["rows"])["opening_credit"] == "1000.00"
    file_id, job_id, _ = await _prepare(maker, storage, result, entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.period_quarter is PeriodQuarter.Q3
    assert document.is_cumulative is expected
    warnings = (await _job(maker, job_id)).metrics[MAPPING_WARNINGS_KEY]
    if expected:
        assert len(warnings) == 1
        assert "нарастающая по данным Q1" in warnings[0]
        assert "sheet:Лист_1" in warnings[0]
        assert document.status is DocumentStatus.NEEDS_CHECK
    else:
        assert warnings == []
        assert document.status is DocumentStatus.CONFIRMED  # чистый → автоподтверждение (OQ-157)


@pytest.mark.parametrize(
    ("job_status", "expected"),
    [(RecognitionStatus.done, DocumentStatus.CONFIRMED), (RecognitionStatus.needs_check, DocumentStatus.NEEDS_CHECK)],
)
async def test_clean_result(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
    job_status: RecognitionStatus,
    expected: DocumentStatus,
):
    """«Чистый результат»: задание `done`, шапка = владелец (ИНН), нарушений нет → `CONFIRMED` (автоподтверждение,
    OQ-157), `MATCH`, `UNKNOWN`;
    задание `needs_check` → `NEEDS_CHECK`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000", job_status=job_status)

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.status is expected
    assert document.counterparty_check is DocumentCheckResult.MATCH
    assert document.period_check is DocumentCheckResult.UNKNOWN
    assert document.osv_account_code == "60"
    assert document.period_quarter is PeriodQuarter.Q1
    assert document.is_cumulative is False
    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED


async def test_counterparty_resolved_in_entries(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """R-3: `counterparty_id` — только у `COUNTERPARTY` с `data_type ∈ {BU, SUM}`; `NU` хранится как пришёл, без
    резолва; строка-счёт без контрагента."""
    maker, storage = committed_sessionmaker, FakeStorage()
    async with maker() as session, session.begin():
        lyutik = await make_legal_entity(session, name="ООО «Лютик»", inn="7711111111")
    result = _minimal()
    rows = _osv(result)["rows"]
    rows.append(rows[1] | {"row_number": 3, "data_type": "NU"})
    file_id, _, _ = await _prepare(maker, storage, result)

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    account, bu, nu = await _entries(maker, document.id)
    assert account.group_type is OsvGroupType.ACCOUNT
    assert account.counterparty_id is None
    assert bu.counterparty_id == lyutik.id
    assert bu.counterparty_inn == "7711111111"
    # v5.16: способ сопоставления хранится ровно у сопоставленных строк (здесь — по ИНН)
    assert (bu.counterparty_match_kind, account.counterparty_match_kind, nu.counterparty_match_kind) == (
        CounterpartyMatchKind.INN,
        None,
        None,
    )
    assert nu.data_type is OsvDataType.NU
    assert nu.counterparty_id is None


async def test_result_of_non_current_job_is_ignored(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Review Focus 1: результат задания, которое уже не текущее для файла, не выбирается и файл не трогает."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, old_job_id, _ = await _prepare(maker, storage, _minimal())
    async with maker() as session, session.begin():
        file = await session.get(StoredFile, file_id)
        assert file is not None
        new_job = RecognitionJob(
            file_id=file_id,
            pipeline=(await session.get(RecognitionJob, old_job_id)).pipeline,  # type: ignore[union-attr]
            request_subject="files.tabular_uploaded",
            schema_version="1.0",
            status=RecognitionStatus.processing,
            attempt=2,
        )
        session.add(new_job)
        await session.flush()
        file.current_recognition_job_id = new_job.id
    before = await _file(maker, file_id)

    assert await process_once(maker, storage, _settings()) is False

    after = await _file(maker, file_id)
    assert (after.status, after.updated_at) == (FileStatus.PROCESSING, before.updated_at)
    assert await _documents(maker, file_id) == []
    assert (await _job(maker, old_job_id)).group_schema is None


async def test_job_replaced_while_reading_result(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """Review Focus 1, гонка: пока цикл читал S3, у файла сменилось текущее задание → транзакция маппинга ничего не
    меняет (задание и файл перечитываются под блокировкой)."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
    real_get_bytes = storage.get_bytes

    async def get_bytes_and_replace_job(key: str) -> bytes:
        async with maker() as session, session.begin():
            file = await session.get(StoredFile, file_id)
            job = await session.get(RecognitionJob, job_id)
            assert file is not None and job is not None
            new_job = RecognitionJob(
                file_id=file_id,
                pipeline=job.pipeline,
                request_subject=job.request_subject,
                schema_version="1.0",
                status=RecognitionStatus.queued,
                attempt=2,
            )
            session.add(new_job)
            await session.flush()
            file.current_recognition_job_id = new_job.id
        return await real_get_bytes(key)

    monkeypatch.setattr(storage, "get_bytes", get_bytes_and_replace_job)

    assert await process_once(maker, storage, _settings()) is True

    file = await _file(maker, file_id)
    assert file.status is FileStatus.PROCESSING
    assert await _documents(maker, file_id) == []
    job = await _job(maker, job_id)
    assert job.group_schema is None
    assert MAPPING_ATTEMPTS_KEY not in (job.metrics or {})


async def test_run_mapping_loop_processes_two_candidates(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """`run_mapping_loop`: два кандидата подряд без паузы (`sleep(0)`), затем пауза `MAPPING_POLL_SECONDS`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    first, _, _ = await _prepare(maker, storage, _minimal())
    second, _, _ = await _prepare(maker, storage, _minimal())
    settings = _settings()
    sleeps: list[float] = []

    async def fake_sleep(seconds: float) -> None:
        sleeps.append(seconds)
        if len(sleeps) == 3:  # noqa: PLR2004 - два кандидата и один пустой проход
            raise asyncio.CancelledError

    with pytest.raises(asyncio.CancelledError):
        await run_mapping_loop(maker, storage, settings, sleep=fake_sleep)

    assert sleeps == [0, 0, settings.MAPPING_POLL_SECONDS]
    assert (await _file(maker, first)).status is FileStatus.PROCESSED
    assert (await _file(maker, second)).status is FileStatus.PROCESSED


async def test_run_mapping_loop_pauses_after_retryable_failure(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """R-15: после повторяемой неудачи цикл спит `MAPPING_POLL_SECONDS` (попытки не сгорают за миллисекунды), после
    успешного маппинга — `sleep(0)`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
    real_map_file = rm.map_file
    failures = [RuntimeError("кратковременный сбой БД")]

    async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
        if failures:
            raise failures.pop()
        return await real_map_file(session, **kwargs)

    monkeypatch.setattr(rm, "map_file", flaky_map_file)
    settings = _settings()
    sleeps: list[float] = []

    async def fake_sleep(seconds: float) -> None:
        sleeps.append(seconds)
        if len(sleeps) == 3:  # noqa: PLR2004 - неудача, успех, пустой проход
            raise asyncio.CancelledError

    with pytest.raises(asyncio.CancelledError):
        await run_mapping_loop(maker, storage, settings, sleep=fake_sleep)

    assert sleeps == [settings.MAPPING_POLL_SECONDS, 0, settings.MAPPING_POLL_SECONDS]
    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
    assert (await _job(maker, job_id)).metrics[MAPPING_ATTEMPTS_KEY] == 1


async def test_run_mapping_loop_survives_errors(monkeypatch: pytest.MonkeyPatch):
    """Ошибка прохода логируется, цикл не умирает и ждёт `MAPPING_POLL_SECONDS`."""
    outcomes: list[Callable[[], bool]] = [lambda: (_ for _ in ()).throw(RuntimeError("сбой")), lambda: True]
    sleeps: list[float] = []

    async def fake_process_once(*args: Any, **kwargs: Any) -> bool:
        return outcomes.pop(0)()

    async def fake_sleep(seconds: float) -> None:
        sleeps.append(seconds)
        if not outcomes:
            raise asyncio.CancelledError

    monkeypatch.setattr(rm, "process_once", fake_process_once)
    settings = _settings()
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="ERROR")
    try:
        with pytest.raises(asyncio.CancelledError):
            await run_mapping_loop(None, FakeStorage(), settings, sleep=fake_sleep)  # type: ignore[arg-type]
    finally:
        logger.remove(handler_id)

    assert sleeps == [settings.MAPPING_POLL_SECONDS, 0]
    assert len(records) == 1


# --- автозамена прежних версий (spec primary-files «Автозамена прежних версий», audit-trail; design D-10) ----------

OldDocument = tuple[str, int, PeriodQuarter, DocumentStatus]  # счёт, год, квартал, статус документа прежнего файла
OSV_60_2025_Q1: OldDocument = ("60", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED)


async def _old_file(  # noqa: PLR0913 - каждое поле прежнего файла задаётся тестом явно
    maker: async_sessionmaker[AsyncSession],
    entity_id: uuid.UUID,
    documents: list[OldDocument],
    *,
    version_no: int = 1,
    status: FileStatus = FileStatus.PROCESSED,
    superseded_by: uuid.UUID | None = None,
    counterparty_check: DocumentCheckResult | None = None,
) -> uuid.UUID:
    """Прежний файл ЮЛ с документами ОСВ и строкой ОСВ на каждый документ (история должна сохраниться);
    `counterparty_check` — у всех документов файла."""
    async with maker() as session, session.begin():
        entity = await session.get(LegalEntity, entity_id)
        assert entity is not None
        file, job = await make_stored_file(
            session, entity=entity, status=status, job_status=RecognitionStatus.done, original_name="old.xls"
        )
        file.version_no = version_no
        file.superseded_by_file_id = superseded_by
        for account, year, quarter, document_status in documents:
            document = await make_document(
                session, file=file, job=job, account=account, year=year, quarter=quarter, status=document_status
            )
            document.counterparty_check = counterparty_check
            session.add(
                OsvEntry(
                    recognition_job_id=job.id,
                    document_id=document.id,
                    owner_legal_entity_id=entity_id,
                    row_number=1,
                    level=0,
                    row_value=account,
                )
            )
        return file.id


async def _entity(
    maker: async_sessionmaker[AsyncSession], name: str = OWNER_NAME, inn: str = "7700000000"
) -> uuid.UUID:
    async with maker() as session, session.begin():
        return (await make_legal_entity(session, name=name, inn=inn)).id


async def _any_user(maker: async_sessionmaker[AsyncSession]) -> uuid.UUID:
    async with maker() as session:
        return (await session.execute(select(AppUser.id).order_by(AppUser.email).limit(1))).scalar_one()


async def _replace_events(maker: async_sessionmaker[AsyncSession]) -> list[AuditEvent]:
    async with maker() as session:
        rows = await session.execute(
            select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_REPLACE).order_by(AuditEvent.occurred_at)
        )
        return list(rows.scalars())


async def _active_files(maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID) -> set[uuid.UUID]:
    """Файлы ЮЛ, которые список файлов показывает по умолчанию (без `include_superseded`)."""
    async with maker() as session:
        rows = await session.execute(
            select(StoredFile.id).where(
                StoredFile.legal_entity_id == entity_id, StoredFile.status != FileStatus.SUPERSEDED
            )
        )
        return set(rows.scalars())


@pytest.mark.parametrize(
    ("old_status", "confirmed"),
    [(DocumentStatus.CONFIRMED, "confirmed=true"), (DocumentStatus.RECOGNIZED, "confirmed=false")],
)
async def test_new_osv_60_supersedes_previous_file(
    committed_sessionmaker: async_sessionmaker[AsyncSession], old_status: DocumentStatus, confirmed: str
):
    """«Новая ОСВ 60 за 1 кв. 2025»: прежний файл (документ `CONFIRMED` тоже, решение владельца 05.10.2026) —
    `SUPERSEDED`, `superseded_by_file_id` = новый, у нового `version_no = 2`; `FILE_REPLACE` — системное событие
    (`user_id` NULL, инициатор загрузки — в `description`); документы и строки прежнего файла сохранены."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    uploader = await _any_user(maker)
    old_id = await _old_file(maker, entity_id, [("60", 2025, PeriodQuarter.Q1, old_status)])
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=uploader)

    assert await process_once(maker, storage, _settings()) is True

    old, new = await _file(maker, old_id), await _file(maker, new_id)
    assert old.status is FileStatus.SUPERSEDED
    assert old.superseded_by_file_id == new_id
    assert old.updated_at is not None
    assert (new.status, new.version_no, new.superseded_by_file_id) == (FileStatus.PROCESSED, 2, None)
    assert await _active_files(maker, entity_id) == {new_id}
    [old_document] = await _documents(maker, old_id)  # история: документ и строки прежнего файла на месте
    assert old_document.status is old_status
    assert len(await _entries(maker, old_document.id)) == 1

    [event] = await _replace_events(maker)
    assert event.user_id is None
    assert event.section is AuditSection.PRIMARY_DOCUMENTS
    assert (event.object_type, event.object_id) == ("stored_file", old_id)
    assert event.after_value == {"file_id": str(new_id), "version_no": 2, "unbound_version_ids": []}
    assert event.description is not None
    assert f"user:{uploader}" in event.description
    assert f"OSV 60 2025 Q1 ({old_status.value})" in event.description
    assert confirmed in event.description


@pytest.mark.parametrize(
    "document",
    [
        ("62", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED),  # другой счёт
        ("60", 2024, PeriodQuarter.Q1, DocumentStatus.CONFIRMED),  # тот же счёт, другой год
        ("60", 2025, PeriodQuarter.Q2, DocumentStatus.CONFIRMED),  # тот же счёт, другой квартал
    ],
)
async def test_other_document_is_not_superseded(
    committed_sessionmaker: async_sessionmaker[AsyncSession], document: OldDocument
):
    """Два разных счёта (или периода) не заменяют друг друга: прежний файл `PROCESSED`, у нового `version_no = 1`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_id = await _old_file(maker, entity_id, [document])
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    old, new = await _file(maker, old_id), await _file(maker, new_id)
    assert (old.status, old.superseded_by_file_id) == (FileStatus.PROCESSED, None)
    assert (new.status, new.version_no) == (FileStatus.PROCESSED, 1)
    assert await _replace_events(maker) == []


async def test_other_legal_entity_is_not_superseded(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Тот же документ у другого ЮЛ не затрагивается: замена — только в пределах ЮЛ-владельца."""
    maker, storage = committed_sessionmaker, FakeStorage()
    other_id = await _entity(maker, name="ООО «Другое ЮЛ»", inn="7799999999")
    old_id = await _old_file(maker, other_id, [OSV_60_2025_Q1])
    new_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000")

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, old_id)).status is FileStatus.PROCESSED
    assert (await _file(maker, new_id)).version_no == 1
    assert await _replace_events(maker) == []


FOREIGN_INN = "5643007516"  # ИНН другой организации в шапке (AURORA-1644: скан чужого баланса в карточке ЮЛ)


@pytest.mark.parametrize(
    ("old_check", "new_inn"),
    [
        (DocumentCheckResult.MATCH, FOREIGN_INN),  # новый документ — другой организации
        (DocumentCheckResult.MISMATCH, "7700000000"),  # прежний документ — другой организации
        (DocumentCheckResult.MISMATCH, FOREIGN_INN),  # оба — не организации карточки
    ],
)
async def test_mismatch_pair_is_not_superseded(
    committed_sessionmaker: async_sessionmaker[AsyncSession], old_check: DocumentCheckResult, new_inn: str
):
    """AURORA-1644: пара «новый — прежний» с `counterparty_check = MISMATCH` у любого из двух в автозамене не
    участвует: прежний файл остаётся `PROCESSED` рядом с новым, у нового `version_no = 1`, `FILE_REPLACE` нет;
    в `mapping_warnings` нового файла — предупреждение с именем и версией оставленного файла."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1], counterparty_check=old_check)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    result = _minimal()
    _osv(result)["header"]["organization_inn"] = new_inn
    new_id, new_job_id, _ = await _prepare(maker, storage, result, entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    old, new = await _file(maker, old_id), await _file(maker, new_id)
    assert (old.status, old.superseded_by_file_id) == (FileStatus.PROCESSED, None)
    assert (new.status, new.version_no) == (FileStatus.PROCESSED, 1)
    assert await _active_files(maker, entity_id) == {old_id, new_id}
    assert await _replace_events(maker) == []
    [warning] = [w for w in await _mapping_warnings(maker, new_job_id) if w.startswith("автозамена")]
    assert warning.startswith("автозамена не выполнена: файл «old.xls» (версия 1) оставлен")
    assert "OSV 60 2025 Q1 (CONFIRMED)" in warning


@pytest.mark.parametrize(
    ("old_check", "new_header"),
    [
        (DocumentCheckResult.UNKNOWN, {"organization_inn": "7700000000"}),  # у прежнего шапка пустая
        (DocumentCheckResult.MATCH, {"organization_inn": None, "organization_name": "ООО"}),  # у нового пустая
    ],
)
async def test_unknown_header_still_supersedes(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
    old_check: DocumentCheckResult,
    new_header: dict[str, Any],
):
    """Пустая шапка (`UNKNOWN`) — не несовпадение: пара заменяется как раньше, предупреждения о пропуске нет."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1], counterparty_check=old_check)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    result = _minimal()
    _osv(result)["header"].update(new_header)
    new_id, new_job_id, _ = await _prepare(maker, storage, result, entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    old, new = await _file(maker, old_id), await _file(maker, new_id)
    assert (old.status, old.superseded_by_file_id, new.version_no) == (FileStatus.SUPERSEDED, new_id, 2)
    assert not [w for w in await _mapping_warnings(maker, new_job_id) if w.startswith("автозамена не выполнена")]


async def test_supersede_is_idempotent(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """Сбой после автозамены откатывает и замену, и аудит; повтор маппинга даёт ровно одно `FILE_REPLACE`. Повторный
    вызов `supersede_previous` для того же файла ничего не находит (заменённый уже `SUPERSEDED`)."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1])
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)
    real_map_file = rm.map_file
    failures = [RuntimeError("сбой БД после автозамены")]

    async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
        warnings = await real_map_file(session, **kwargs)
        if failures:
            raise failures.pop()
        return warnings

    monkeypatch.setattr(rm, "map_file", flaky_map_file)

    assert await process_once(maker, storage, _settings()) is False  # повторяемая неудача: всё откачено
    assert (await _file(maker, old_id)).status is FileStatus.PROCESSED
    assert await _replace_events(maker) == []

    assert await process_once(maker, storage, _settings()) is True
    assert (await _file(maker, old_id)).status is FileStatus.SUPERSEDED
    assert len(await _replace_events(maker)) == 1

    async with maker() as session, session.begin():
        new = await session.get(StoredFile, new_id)
        assert new is not None
        documents = list((await session.execute(select(Document).where(Document.stored_file_id == new_id))).scalars())
        replaced = await supersede_previous(
            session, new_file=new, new_documents=documents, uploader_id=None, now=_now()
        )
    assert replaced == []
    assert len(await _replace_events(maker)) == 1
    assert (await _file(maker, new_id)).version_no == 2  # noqa: PLR2004 - версия не растёт от повторного вызова


async def test_version_chain(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Цепочка v1 → v2 → v3: третий файл заменяет только действующий v2 и получает `version_no = 3`; v1 остаётся
    заменённым файлом v2."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    v2_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1], version_no=2)
    v1_id = await _old_file(
        maker, entity_id, [OSV_60_2025_Q1], version_no=1, status=FileStatus.SUPERSEDED, superseded_by=v2_id
    )
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    v3_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    v1, v2, v3 = await _file(maker, v1_id), await _file(maker, v2_id), await _file(maker, v3_id)
    assert (v1.status, v1.superseded_by_file_id) == (FileStatus.SUPERSEDED, v2_id)
    assert (v2.status, v2.superseded_by_file_id) == (FileStatus.SUPERSEDED, v3_id)
    assert v3.version_no == 3  # noqa: PLR2004 - третья версия
    [event] = await _replace_events(maker)
    assert event.object_id == v2_id
    assert event.after_value == {"file_id": str(v3_id), "version_no": 3, "unbound_version_ids": []}


async def _consolidation_id(
    maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID, quarter: PeriodQuarter
) -> uuid.UUID:
    """Консолидация ГК этого ЮЛ за 2025/quarter (без версий: для наследования они не нужны).

    ЮЛ без ГК получает ГК (консолидация требует группу) и привязывается к ней."""
    async with maker() as session, session.begin():
        entity = await session.get(LegalEntity, entity_id)
        assert entity is not None
        if entity.company_group_id is None:
            group = await make_company_group(session, name="ГК Наследование")
            entity.company_group_id = group.id
        consolidation = Consolidation(
            company_group_id=entity.company_group_id,
            year=2025,
            quarter=quarter,
            reporting_date=reporting_date(2025, quarter),
            created_at=_now(),
        )
        session.add(consolidation)
        await session.flush()
        return consolidation.id


async def _set_consolidation(
    maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID, consolidation_id: uuid.UUID | None
) -> None:
    async with maker() as session, session.begin():
        await session.execute(
            update(StoredFile).where(StoredFile.id == file_id).values(consolidation_id=consolidation_id)
        )


async def test_supersede_inherits_consolidation_from_first_with_value(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
) -> None:
    """Спека «Новый файл без контекста наследует консолидацию» + Review Focus 5: два заменяемых файла, у первого по `id`
    `consolidation_id` пуст, у второго задан — новый получает значение второго."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    c1 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q1)
    first_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1])
    second_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1])
    empty_id, filled_id = sorted([first_id, second_id])
    await _set_consolidation(maker, filled_id, c1)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, new_id)).consolidation_id == c1
    assert (await _file(maker, empty_id)).status is FileStatus.SUPERSEDED
    assert (await _file(maker, filled_id)).status is FileStatus.SUPERSEDED


async def test_supersede_keeps_own_consolidation(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None:
    """Спека «Новый файл с контекстом сохраняет свой»: старый в C1, новый загружен в C2 — остаётся C2."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    c1 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q1)
    c2 = await _consolidation_id(maker, entity_id, PeriodQuarter.Q2)
    old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1])
    await _set_consolidation(maker, old_id, c1)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)
    await _set_consolidation(maker, new_id, c2)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, new_id)).consolidation_id == c2
    assert (await _file(maker, old_id)).status is FileStatus.SUPERSEDED


async def test_one_event_per_replaced_file(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Новый файл с ОСВ 60 и 62 заменяет прежний файл с теми же двумя документами одним событием `FILE_REPLACE`,
    в `description` — оба документа."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_id = await _old_file(
        maker,
        entity_id,
        [OSV_60_2025_Q1, ("62", 2025, PeriodQuarter.Q1, DocumentStatus.NEEDS_CHECK)],
    )
    result = _minimal()
    second = copy.deepcopy(_osv(result)) | {"locator": "sheet:Лист_2"}
    second["header"]["account_code_raw"] = "62"
    result["documents"].append(second)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, result, entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, old_id)).superseded_by_file_id == new_id
    assert len(await _documents(maker, new_id)) == 2  # noqa: PLR2004 - два листа, два счёта
    [event] = await _replace_events(maker)
    assert event.object_id == old_id
    assert event.description is not None
    assert "OSV 60 2025 Q1 (CONFIRMED)" in event.description
    assert "OSV 62 2025 Q1 (NEEDS_CHECK)" in event.description
    assert "confirmed=true" in event.description


async def test_new_file_supersedes_several_files(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Один новый файл (ОСВ 60, 62, 66) заменяет три разных прежних файла: событие `FILE_REPLACE` на каждый заменённый
    файл, у нового `version_no` = МАКСИМУМ заменённых + 1. Версии раздаются по возрастанию `id` как 1, 3, 2: максимум —
    в середине порядка блокировки, поэтому ни первый, ни последний заблокированный файл его не дают."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    old_ids = [
        await _old_file(maker, entity_id, [(account, 2025, PeriodQuarter.Q1, DocumentStatus.RECOGNIZED)])
        for account in ("60", "62", "66")
    ]
    ordered = sorted(old_ids)
    versions = dict(zip(ordered, (1, 3, 2), strict=True))
    async with maker() as session, session.begin():
        for file_id, version_no in versions.items():
            file = await session.get(StoredFile, file_id)
            assert file is not None
            file.version_no = version_no
    result = _minimal()
    for number, account in enumerate(("62", "66"), start=2):
        sheet = copy.deepcopy(_osv(result)) | {"locator": f"sheet:Лист_{number}"}
        sheet["header"]["account_code_raw"] = account
        result["documents"].append(sheet)
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    new_id, _, _ = await _prepare(maker, storage, result, entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    expected_version = 4  # max(1, 3, 2) + 1
    for old_id in old_ids:
        old = await _file(maker, old_id)
        assert (old.status, old.superseded_by_file_id) == (FileStatus.SUPERSEDED, new_id)
    assert (await _file(maker, new_id)).version_no == expected_version
    events = await _replace_events(maker)
    assert sorted(event.object_id for event in events) == ordered  # type: ignore[type-var]
    for event in events:
        assert event.user_id is None
        assert event.after_value == {
            "file_id": str(new_id),
            "version_no": expected_version,
            "unbound_version_ids": [],
        }


async def test_replaced_document_is_unbound_from_versions_in_work(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """primary-files «Заменённый документ был в версии»: `CONFIRMED` ОСВ 60 за 2024 Q2 привязана к версии 2024 Q3 в
    `DATA_CORRECTION` и к аннулированной версии; автозамена отвязывает документ только от версии «в работе» (статус
    прежний), `FILE_REPLACE.after_value.unbound_version_ids` — эта версия. Привязка другого документа заменённого
    файла (ОСВ 62, ключ замены не совпал) тоже снимается: заменён весь файл."""
    maker = committed_sessionmaker
    async with maker() as session, session.begin():
        initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one()
        group = await make_company_group(session, name="ГК Автозамена")
        entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000", group=group)
        other = await make_legal_entity(session, name="ООО Второе", group=group)
        consolidation, in_work = await make_consolidation(
            session,
            group=group,
            year=2024,
            quarter=PeriodQuarter.Q3,
            members=[entity, other],
            initiator=initiator,
            status=ConsolidationStatus.DATA_CORRECTION,
        )
        cancelled = await make_version(
            session,
            consolidation=consolidation,
            version_no=2,
            members=[entity, other],
            initiator=initiator,
            status=ConsolidationStatus.CANCELLED,
        )
        old_file, old_job = await make_stored_file(
            session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
        )
        old_60 = await make_document(
            session,
            file=old_file,
            job=old_job,
            account="60",
            year=2024,
            quarter=PeriodQuarter.Q2,
            status=DocumentStatus.CONFIRMED,
            is_cumulative=False,
        )
        old_62 = await make_document(
            session,
            file=old_file,
            job=old_job,
            account="62",
            year=2024,
            quarter=PeriodQuarter.Q3,
            status=DocumentStatus.CONFIRMED,
            is_cumulative=True,
        )
        for version in (in_work, cancelled):
            for document in (old_60, old_62):
                session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
        new_file, new_job = await make_stored_file(
            session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
        )
        new_60 = await make_document(
            session,
            file=new_file,
            job=new_job,
            account="60",
            year=2024,
            quarter=PeriodQuarter.Q2,
            status=DocumentStatus.NEEDS_CHECK,
            is_cumulative=False,
        )
        ids = (in_work.id, cancelled.id, old_file.id, old_60.id, old_62.id)

    async with maker() as session, session.begin():
        new = await session.get(StoredFile, new_file.id)
        assert new is not None
        replaced = await supersede_previous(session, new_file=new, new_documents=[new_60], uploader_id=None, now=_now())
    in_work_id, cancelled_id, old_file_id, old_60_id, old_62_id = ids

    assert [file.id for file in replaced] == [old_file_id]
    async with maker() as session:

        def bound(version_id: uuid.UUID) -> Any:
            return select(ConsolidationDocument.document_id).where(
                ConsolidationDocument.consolidation_version_id == version_id
            )

        assert set((await session.execute(bound(in_work_id))).scalars()) == set()
        assert set((await session.execute(bound(cancelled_id))).scalars()) == {old_60_id, old_62_id}
        statuses = dict(
            (await session.execute(select(ConsolidationVersion.id, ConsolidationVersion.status))).tuples().all()
        )
        assert statuses == {
            in_work_id: ConsolidationStatus.DATA_CORRECTION,
            cancelled_id: ConsolidationStatus.CANCELLED,
        }
    [event] = await _replace_events(maker)
    assert event.object_id == old_file_id
    assert event.after_value == {"file_id": str(new_file.id), "version_no": 2, "unbound_version_ids": [str(in_work_id)]}


# --- OQ-154: предупреждение о документах заменённого файла, которых нет в новом -----------------------------------


STATEMENT_MINIMAL = API_DIR / "contracts" / "examples" / "v1" / "result-statement-minimal.json"


def _statement_result(*document_types: str) -> dict[str, Any]:
    """Результат с формами `document_types` за 30.09.2025 (Q3 2025), копии `result-statement-minimal.json`."""
    result = json.loads(STATEMENT_MINIMAL.read_text("utf-8"))
    template = result["documents"][0]
    result["documents"] = []
    for number, document_type in enumerate(document_types, start=1):
        document = copy.deepcopy(template) | {"document_type": document_type, "locator": f"pages:{number}"}
        result["documents"].append(document)
    return result


async def _mapping_warnings(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> list[str]:
    job = await _job(maker, job_id)
    assert isinstance(job.metrics, dict)
    warnings: list[str] = job.metrics[MAPPING_WARNINGS_KEY]
    return warnings


async def test_replace_warns_about_dropped_confirmed_balance(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """OQ-154: книга «ББ + ОФР» за Q3 2025, ББ подтверждён и привязан к версии «в работе»; отдельный файл только с ОФР
    того же периода заменяет всю книгу, а в `mapping_warnings` нового задания — имя прежнего файла, выпавший
    `BALANCE 2025 Q3 (CONFIRMED)` и версия, с которой снята привязка. Замена и статус документа ОФР не меняются."""
    maker, storage = committed_sessionmaker, FakeStorage()
    book_id, book_job_id, entity_id = await _prepare(
        maker, storage, _statement_result("BALANCE", "INCOME"), owner_inn="7700000000"
    )
    assert await process_once(maker, storage, _settings()) is True
    assert await _mapping_warnings(maker, book_job_id) == []  # книга ничего не заменяла: предупреждений нет
    async with maker() as session, session.begin():
        initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one()
        entity = await session.get(LegalEntity, entity_id)
        assert entity is not None
        group = await make_company_group(session, name="ГК Книга")
        entity.company_group_id = group.id
        _, version = await make_consolidation(
            session,
            group=group,
            year=2025,
            quarter=PeriodQuarter.Q3,
            members=[entity],
            initiator=initiator,
            status=ConsolidationStatus.DATA_CORRECTION,
        )
        documents = {
            doc.document_type: doc
            for doc in (await session.execute(select(Document).where(Document.stored_file_id == book_id))).scalars()
        }
        assert set(documents) == {DocumentType.BALANCE, DocumentType.INCOME}
        documents[DocumentType.BALANCE].status = DocumentStatus.CONFIRMED
        session.add(
            ConsolidationDocument(consolidation_version_id=version.id, document_id=documents[DocumentType.BALANCE].id)
        )
        version_id = version.id
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)

    new_id, new_job_id, _ = await _prepare(maker, storage, _statement_result("INCOME"), entity=entity)
    assert await process_once(maker, storage, _settings()) is True

    [warning] = await _mapping_warnings(maker, new_job_id)
    book_name = (await _file(maker, book_id)).original_name
    assert warning.startswith(f"автозамена: файл «{book_name}» (версия 1) заменён этим файлом")
    assert "BALANCE 2025 Q3 (CONFIRMED)" in warning
    assert "INCOME" not in warning  # ОФР есть в новом файле: не выпал
    assert "ГК Книга" in warning and "2025 Q3, версия 1" in warning
    book = await _file(maker, book_id)
    assert (book.status, book.superseded_by_file_id) == (FileStatus.SUPERSEDED, new_id)  # поведение замены прежнее
    async with maker() as session:
        bound = (
            await session.execute(
                select(ConsolidationDocument).where(ConsolidationDocument.consolidation_version_id == version_id)
            )
        ).all()
    assert bound == []
    [event] = await _replace_events(maker)
    assert event.after_value is not None and event.after_value["unbound_version_ids"] == [str(version_id)]


async def test_replace_without_dropped_documents_has_no_warning(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """Новый файл содержит ключи всех документов прежнего — предупреждения нет (обычная автозамена)."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id = await _entity(maker)
    await _old_file(maker, entity_id, [OSV_60_2025_Q1])
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
    _, new_job_id, _ = await _prepare(maker, storage, _minimal(), entity=entity)

    assert await process_once(maker, storage, _settings()) is True

    assert await _mapping_warnings(maker, new_job_id) == []


async def test_supersede_previous_reports_dropped_documents_without_changing_replace(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """`supersede_previous(warnings=…)`: по предупреждению на заменённый файл; выпавшие — и подтверждённый, и нет, и
    документ другого периода; версия «не в работе» в предупреждении не фигурирует (привязка не снималась)."""
    maker = committed_sessionmaker
    async with maker() as session, session.begin():
        initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one()
        group = await make_company_group(session, name="ГК Выпавшие")
        entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000", group=group)
        consolidation, in_work = await make_consolidation(
            session,
            group=group,
            year=2025,
            quarter=PeriodQuarter.Q1,
            members=[entity],
            initiator=initiator,
            status=ConsolidationStatus.DATA_CORRECTION,
        )
        cancelled = await make_version(
            session,
            consolidation=consolidation,
            version_no=2,
            members=[entity],
            initiator=initiator,
            status=ConsolidationStatus.CANCELLED,
        )
        old_file, old_job = await make_stored_file(
            session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
        )
        await make_document(session, file=old_file, job=old_job, account="60", status=DocumentStatus.CONFIRMED)
        old_62 = await make_document(session, file=old_file, job=old_job, account="62", status=DocumentStatus.CONFIRMED)
        old_66 = await make_document(
            session, file=old_file, job=old_job, account="66", status=DocumentStatus.NEEDS_CHECK
        )
        for version in (in_work, cancelled):
            for document in (old_62, old_66):
                session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
        new_file, new_job = await make_stored_file(
            session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
        )
        new_60 = await make_document(session, file=new_file, job=new_job, account="60")
        new_file_id, in_work_id = new_file.id, in_work.id

    warnings: list[str] = []
    async with maker() as session, session.begin():
        new = await session.get(StoredFile, new_file_id)
        assert new is not None
        replaced = await supersede_previous(
            session, new_file=new, new_documents=[new_60], uploader_id=None, now=_now(), warnings=warnings
        )

    assert len(replaced) == 1
    [warning] = warnings
    assert "OSV 62 2025 Q1 (CONFIRMED)" in warning and "OSV 66 2025 Q1 (NEEDS_CHECK)" in warning
    assert "OSV 60" not in warning  # совпал по ключу — не выпал
    assert "«ГК Выпавшие» 2025 Q1, версия 1" in warning
    assert "версия 2" not in warning.split("привязки сняты", 1)[1]  # CANCELLED-версия не затронута
    [event] = await _replace_events(maker)
    assert event.after_value is not None and event.after_value["unbound_version_ids"] == [str(in_work_id)]


# --- osv-result-mapping «Документ NEEDS_CHECK переводит версию в корректировку» (D-6) ------------------------------


async def _version_in(
    maker: async_sessionmaker[AsyncSession],
    *,
    status: ConsolidationStatus = ConsolidationStatus.DOCUMENTS_PROCESSING,
    quarter: PeriodQuarter = PeriodQuarter.Q3,
) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
    """ГК с ЮЛ-владельцем (ИНН как в результате) и вторым ЮЛ; версия `quarter` 2025; `(entity, version, user)`."""
    async with maker() as session, session.begin():
        initiator = (await session.execute(select(AppUser).order_by(AppUser.email).limit(1))).scalar_one()
        group = await make_company_group(session, name="ГК Маппинг версии")
        owner = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000", group=group)
        other = await make_legal_entity(session, name="ООО Второе", group=group)
        _, version = await make_consolidation(
            session, group=group, year=2025, quarter=quarter, members=[owner, other], initiator=initiator, status=status
        )
        return owner.id, version.id, initiator.id


async def _load_entity(maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID) -> LegalEntity:
    async with maker() as session:
        entity = await session.get(LegalEntity, entity_id)
        assert entity is not None
        return entity


@pytest.mark.parametrize(
    ("job_status", "moved"),
    [(RecognitionStatus.needs_check, True), (RecognitionStatus.done, False)],  # NEEDS_CHECK / CONFIRMED (чистый)
)
async def test_mapping_moves_documents_processing_version_only_for_needs_check(
    committed_sessionmaker: async_sessionmaker[AsyncSession], job_status: RecognitionStatus, moved: bool
):
    """«Предупреждение маппинга двигает статус» и «Документ RECOGNIZED»: ОСВ 60 за 2025 Q1 (квартальная) подходит версии
    2025 Q3; `NEEDS_CHECK` → `DATA_CORRECTION`, запись истории с `changed_by` = загрузивший, привязок нет; чистый
    документ → `CONFIRMED`, привязка и тот же переход (OQ-157)."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, user_id = await _version_in(maker)
    file_id, _, _ = await _prepare(
        maker,
        storage,
        _minimal(),
        entity=await _load_entity(maker, entity_id),
        job_status=job_status,
        uploaded_by=user_id,
    )

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.status is (DocumentStatus.NEEDS_CHECK if moved else DocumentStatus.CONFIRMED)
    async with maker() as session:
        version = await session.get(ConsolidationVersion, version_id)
        assert version is not None
        history = await _history_of(maker, version_id)
        bindings = (await session.execute(select(func.count()).select_from(ConsolidationDocument))).scalar_one()
    if moved:
        assert bindings == 0
        assert version.status is ConsolidationStatus.DATA_CORRECTION
        last = history[-1]
        assert (last.from_status, last.to_status, last.changed_by) == (
            ConsolidationStatus.DOCUMENTS_PROCESSING,
            ConsolidationStatus.DATA_CORRECTION,
            user_id,
        )
    else:
        # чистый документ подтверждён системой (OQ-157): привязан, версия перешла от загрузившего с комментарием
        assert bindings == 1
        assert version.status is ConsolidationStatus.DATA_CORRECTION
        last = history[-1]
        assert (last.from_status, last.to_status, last.changed_by, last.comment) == (
            ConsolidationStatus.DOCUMENTS_PROCESSING,
            ConsolidationStatus.DATA_CORRECTION,
            user_id,
            "автоподтверждение документа OSV 60",
        )


async def test_mapping_does_not_move_initiated_version(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Только `DOCUMENTS_PROCESSING` реагирует на маппинг (spec): версия `INITIATED` остаётся."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, user_id = await _version_in(maker, status=ConsolidationStatus.INITIATED)
    await _prepare(
        maker,
        storage,
        _minimal(),
        entity=await _load_entity(maker, entity_id),
        job_status=RecognitionStatus.needs_check,
        uploaded_by=user_id,
    )

    assert await process_once(maker, storage, _settings()) is True

    async with maker() as session:
        version = await session.get(ConsolidationVersion, version_id)
        assert version is not None
        assert version.status is ConsolidationStatus.INITIATED


async def _history_of(
    maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID
) -> list[ConsolidationStatusHistory]:
    async with maker() as session:
        stmt = (
            select(ConsolidationStatusHistory)
            .where(ConsolidationStatusHistory.version_id == version_id)
            .order_by(ConsolidationStatusHistory.changed_at)
        )
        return list((await session.execute(stmt)).scalars())


async def _bound_ids(maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID) -> set[uuid.UUID]:
    async with maker() as session:
        stmt = select(ConsolidationDocument.document_id).where(
            ConsolidationDocument.consolidation_version_id == version_id
        )
        return set((await session.execute(stmt)).scalars())


async def test_clean_document_is_auto_confirmed_bound_and_moves_version(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """osv-result-mapping «Привязка и переход версии»: чистая ОСВ 60 за 2025 Q1 → `CONFIRMED` системой, привязана к
    версии 2025 Q3 (`DOCUMENTS_PROCESSING`), переход в `DATA_CORRECTION` от загрузившего с комментарием."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, user_id = await _version_in(maker)
    file_id, _, _ = await _prepare(
        maker, storage, _minimal(), entity=await _load_entity(maker, entity_id), uploaded_by=user_id
    )

    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert (document.status, document.confirmed_by) == (DocumentStatus.CONFIRMED, None)
    assert document.confirmed_at is not None
    assert await _bound_ids(maker, version_id) == {document.id}
    last = (await _history_of(maker, version_id))[-1]
    assert (last.to_status, last.changed_by, last.comment) == (
        ConsolidationStatus.DATA_CORRECTION,
        user_id,
        "автоподтверждение документа OSV 60",
    )
    async with maker() as session:
        events = (
            (
                await session.execute(
                    select(AuditEvent).where(
                        AuditEvent.action == AuditAction.DOCUMENT_CONFIRM, AuditEvent.object_id == document.id
                    )
                )
            )
            .scalars()
            .all()
        )
    assert [(event.user_id, event.description) for event in events] == [(None, "автоподтверждение: без замечаний")]


async def test_mixed_file_confirms_clean_and_moves_version_once(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """Review Focus 1: в файле чистый и `NEEDS_CHECK` документы — версия получает ровно один переход."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, user_id = await _version_in(maker)
    result = _minimal()
    second = copy.deepcopy(result["documents"][0])
    second["header"]["account_code_raw"] = "62"
    second["locator"] = "sheet:Лист_2"
    second["checks"] = {"total_row_matches_root": False}  # нарушение проверок → NEEDS_CHECK
    result["documents"].append(second)
    file_id, _, _ = await _prepare(
        maker, storage, result, entity=await _load_entity(maker, entity_id), uploaded_by=user_id
    )

    assert await process_once(maker, storage, _settings()) is True

    statuses = sorted((d.osv_account_code, d.status) for d in await _documents(maker, file_id))
    assert statuses == [("60", DocumentStatus.CONFIRMED), ("62", DocumentStatus.NEEDS_CHECK)]
    history = await _history_of(maker, version_id)
    assert [h.to_status for h in history].count(ConsolidationStatus.DATA_CORRECTION) == 1
    # переход сделал именно автоподтверждение (T4.2): при перестановке с on_documents_mapped комментарий иной
    assert history[-1].to_status is ConsolidationStatus.DATA_CORRECTION
    assert history[-1].comment == "автоподтверждение документа OSV 60"
    clean = next(d for d in await _documents(maker, file_id) if d.osv_account_code == "60")
    assert await _bound_ids(maker, version_id) == {clean.id}  # NEEDS_CHECK-документ 62 не привязан


async def test_supersede_rebinds_auto_confirmed_document(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Review Focus 2: новая чистая версия файла того же ключа заменяет привязку старой."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, user_id = await _version_in(maker)
    entity = await _load_entity(maker, entity_id)
    first_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id)
    assert await process_once(maker, storage, _settings()) is True
    [first] = await _documents(maker, first_id)
    assert await _bound_ids(maker, version_id) == {first.id}

    second_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id)
    assert await process_once(maker, storage, _settings()) is True

    [second] = await _documents(maker, second_id)
    assert second.status is DocumentStatus.CONFIRMED
    assert await _bound_ids(maker, version_id) == {second.id}
    assert (await _file(maker, first_id)).status is FileStatus.SUPERSEDED


async def test_auto_confirm_rolls_back_with_mapping(
    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
):
    """Review Focus 3: сбой после автоподтверждения откатывает и его; повтор — одно событие `DOCUMENT_CONFIRM`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, _, user_id = await _version_in(maker)
    file_id, job_id, _ = await _prepare(
        maker, storage, _minimal(), entity=await _load_entity(maker, entity_id), uploaded_by=user_id
    )
    original = consolidation_documents.on_documents_mapped
    calls = {"n": 0}

    async def flaky(*args: Any, **kwargs: Any) -> list[uuid.UUID]:
        calls["n"] += 1
        if calls["n"] == 1:
            raise RuntimeError("сбой после автоподтверждения")
        return await original(*args, **kwargs)

    monkeypatch.setattr(consolidation_documents, "on_documents_mapped", flaky)

    assert await process_once(maker, storage, _settings()) is False  # повторяемая неудача (R-15)
    assert await _documents(maker, file_id) == []  # маппинг вместе с автоподтверждением откатился целиком
    assert await process_once(maker, storage, _settings()) is True

    [document] = await _documents(maker, file_id)
    assert document.status is DocumentStatus.CONFIRMED
    async with maker() as session:
        count = (
            await session.execute(
                select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.DOCUMENT_CONFIRM)
            )
        ).scalar_one()
    assert count == 1


async def test_mapping_without_uploader_skips_transition_without_error(
    committed_sessionmaker: async_sessionmaker[AsyncSession],
):
    """`uploaded_by` NULL: маппинг завершается (файл `PROCESSED`), версия не двигается, причина — в логе."""
    maker, storage = committed_sessionmaker, FakeStorage()
    entity_id, version_id, _ = await _version_in(maker)
    file_id, _, _ = await _prepare(
        maker,
        storage,
        _minimal(),
        entity=await _load_entity(maker, entity_id),
        job_status=RecognitionStatus.needs_check,
        uploaded_by=None,
    )
    messages: list[str] = []
    sink = logger.add(lambda message: messages.append(str(message)), level="INFO")
    try:
        assert await process_once(maker, storage, _settings()) is True
    finally:
        logger.remove(sink)

    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
    async with maker() as session:
        version = await session.get(ConsolidationVersion, version_id)
        assert version is not None
        assert version.status is ConsolidationStatus.DOCUMENTS_PROCESSING
    assert any(str(file_id) in message and "uploaded_by" in message for message in messages)


# --- audit-trail: обмен и маппинг не пишут audit_event (ASSUMPTION(OQ-55)) ----------------------------------------


async def _audit_count(maker: async_sessionmaker[AsyncSession]) -> int:
    async with maker() as session:
        return (await session.execute(select(func.count()).select_from(AuditEvent))).scalar_one()


async def test_exchange_and_mapping_write_no_audit_events(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """spec audit-trail «События файлов и документов»: смена статусов задания и файла (ack, результат, маппинг) и
    автозамена без заменяемых файлов `audit_event` не пишут: путь `UPLOADED → PROCESSING → PROCESSED` для документа
    `NEEDS_CHECK` (у владельца нет ИНН) оставляет таблицу аудита неизменной. Автоподтверждение чистого документа пишет
    `DOCUMENT_CONFIRM` — это проверяет отдельный тест."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(
        maker, storage, _minimal(), job_status=RecognitionStatus.published, file_status=FileStatus.UPLOADED
    )
    before = await _audit_count(maker)

    await handle_message(
        fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), session_factory=maker
    )
    assert (await _file(maker, file_id)).status is FileStatus.PROCESSING
    await handle_message(fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), session_factory=maker)
    assert (await _job(maker, job_id)).status is RecognitionStatus.done
    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
    [document] = await _documents(maker, file_id)
    assert document.status is DocumentStatus.NEEDS_CHECK
    assert await _audit_count(maker) == before


async def test_watchdog_timeout_writes_no_audit_event(committed_sessionmaker: async_sessionmaker[AsyncSession]):
    """Путь ошибки: вахтёр переводит задание в `timeout`, файл — в `ERROR`, `audit_event` не пишется."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal(), job_status=RecognitionStatus.processing)
    async with maker() as session, session.begin():
        job = await session.get(RecognitionJob, job_id)
        assert job is not None
        job.reply_deadline = _now() - timedelta(minutes=1)
    before = await _audit_count(maker)

    assert await expire_deadlines(maker, now=_now) >= 1

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    assert (await _job(maker, job_id)).status is RecognitionStatus.timeout
    assert await _audit_count(maker) == before


# --- v5.19: поля 1.x, file_ordinal, вторая форма, MAPPING_WARNING (план 3, A9) ---------------------------------------


def _osv_sheet(account: str, locator: str) -> dict[str, Any]:
    sheet = copy.deepcopy(_osv(_minimal()))
    sheet["header"]["account_code_raw"] = account
    sheet["locator"] = locator
    return sheet


def _form(document_type: str, locator: str, *, report_date: str = "2025-09-30") -> dict[str, Any]:
    form = copy.deepcopy(json.loads(STATEMENT_MINIMAL.read_text("utf-8"))["documents"][0])
    form["document_type"] = document_type
    form["locator"] = locator
    form["header"]["report_date"] = report_date
    return form


async def _map_dict(
    session: AsyncSession, documents: list[dict[str, Any]], *, owner_inn: str = "7700000000"
) -> tuple[StoredFile, list[Document]]:
    """`map_file` на результате 1.x из словарей `documents`; документы файла по `file_ordinal`."""
    entity = await make_legal_entity(session, name=OWNER_NAME, inn=owner_inn)
    file, job = await make_stored_file(
        session, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.done
    )
    payload = _minimal() | {"job_id": str(job.id), "file_id": str(file.id), "documents": documents}
    result = RecognitionResult.model_validate(payload)
    chart = {code: key for code, key in await session.execute(select(ChartOfAccount.code, ChartOfAccount.id))}
    await rm.map_file(
        session, job=job, file=file, result=result, chart=chart, index=await CounterpartyIndex.load(session), now=_now()
    )
    stored = (
        (
            await session.execute(
                select(Document).where(Document.stored_file_id == file.id).order_by(Document.file_ordinal)
            )
        )
        .scalars()
        .all()
    )
    return file, list(stored)


async def _issues(session: AsyncSession, document: Document) -> list[DocumentIssue]:
    stmt = select(DocumentIssue).where(DocumentIssue.document_id == document.id)
    return list((await session.execute(stmt)).scalars())


async def test_two_osv_sheets_file_ordinals(seeded: AsyncSession) -> None:
    file, documents = await _map_dict(seeded, [_osv_sheet("60", "sheet:60"), _osv_sheet("62", "sheet:62")])
    assert [(d.osv_account_code, d.file_ordinal) for d in documents] == [("60", 0), ("62", 1)]
    await seeded.refresh(file, ["status"])
    assert file.status is FileStatus.PROCESSED


async def test_pdf_balance_income_1x_ordinals(seeded: AsyncSession) -> None:
    _, documents = await _map_dict(seeded, [_form("BALANCE", "pages:1"), _form("INCOME", "pages:2")])
    assert [(d.document_type, d.file_ordinal) for d in documents] == [
        (DocumentType.BALANCE, 0),
        (DocumentType.INCOME, 1),
    ]


async def test_duplicate_osv_sheet_takes_no_ordinal(seeded: AsyncSession) -> None:
    _, documents = await _map_dict(
        seeded, [_osv_sheet("60", "sheet:a"), _osv_sheet("60", "sheet:b"), _form("BALANCE", "sheet:ББ")]
    )
    assert [(d.document_type, d.file_ordinal) for d in documents] == [(DocumentType.OSV, 0), (DocumentType.BALANCE, 2)]


async def test_disclosure_ordinal_is_first_in_documents(seeded: AsyncSession) -> None:
    """P3-6: объединённый `DISCLOSURE` — индекс первого такого документа в `documents[]`."""
    bare = {"header": {}, "columns": [], "lines": []}
    disclosure = {"document_type": "DISCLOSURE", "locator": "pages:3", **bare}
    other = {"document_type": "OTHER", "locator": "pages:9", **bare}
    _, documents = await _map_dict(
        seeded, [_form("BALANCE", "pages:1"), disclosure, other, disclosure | {"locator": "pages:4"}]
    )
    assert {(d.document_type, d.file_ordinal) for d in documents} == {
        (DocumentType.BALANCE, 0),
        (DocumentType.DISCLOSURE, 1),
        (DocumentType.OTHER, 2),
    }


async def test_1x_line_fields(seeded: AsyncSession) -> None:
    form = _form("BALANCE", "sheet:ББ")
    form["lines"] = [
        {
            "line_code": code,
            "line_name": f"строка {code}",
            "values": [
                {
                    "column_index": 0,
                    "amount": "10",
                    "string_value": "10,0",
                    "confidence": confidence,
                    "is_recognized": True,
                }
            ],
        }
        for code, confidence in (("1100", 100), ("1200", 80), ("1600", 100))
    ]
    _, [balance] = await _map_dict(seeded, [form])
    meta = (
        await seeded.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == balance.id))
    ).scalar_one()
    lines = (
        (
            await seeded.execute(
                select(FinancialStatementLine)
                .where(FinancialStatementLine.meta_id == meta.id)
                .order_by(FinancialStatementLine.row_index)
            )
        )
        .scalars()
        .all()
    )
    assert [line.row_index for line in lines] == [0, 1, 2]
    assert all(line.line_status.value == "ok" and line.source_ref is None for line in lines)
    assert all(line.line_code_original_value == line.line_code and line.needs_review is False for line in lines)
    values = {
        line.line_code: (
            await seeded.execute(select(FinancialStatementValue).where(FinancialStatementValue.line_id == line.id))
        )
        .scalars()
        .all()
        for line in lines
    }
    flags = {code: {v.needs_review for v in cells if v.confidence is not None} for code, cells in values.items()}
    assert flags == {"1100": {False}, "1200": {True}, "1600": {False}}
    assert {v.source_text for v in values["1100"] if v.string_value} == {"10,0"}


async def test_1x_duplicate_code_keeps_first(seeded: AsyncSession) -> None:
    form = _form("INCOME", "sheet:ОФР")
    form["lines"] = [
        {"line_code": "2110", "line_name": "а", "values": []},
        {"line_code": "2410", "line_name": "б", "values": []},
        {"line_code": "2410", "line_name": "повтор", "values": []},
    ]
    _, [income] = await _map_dict(seeded, [form])
    meta = (
        await seeded.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == income.id))
    ).scalar_one()
    rows = (
        await seeded.execute(
            select(FinancialStatementLine.row_index, FinancialStatementLine.line_code)
            .where(FinancialStatementLine.meta_id == meta.id)
            .order_by(FinancialStatementLine.row_index)
        )
    ).all()
    assert [tuple(row) for row in rows] == [(0, "2110"), (1, "2410")]
    messages = [issue.message for issue in await _issues(seeded, income) if issue.code == "MAPPING_WARNING"]
    assert any("2410" in message and "повторяется" in message for message in messages)


async def test_1x_second_form_same_period_stored(seeded: AsyncSession) -> None:
    """Review Focus 1, P3-4 (Р-10/Р-18): две формы одного типа и периода хранятся, обе `NEEDS_CHECK`."""
    file, documents = await _map_dict(seeded, [_form("BALANCE", "sheet:а"), _form("BALANCE", "sheet:б")])
    assert [d.file_ordinal for d in documents] == [0, 1]
    assert all(d.status is DocumentStatus.NEEDS_CHECK for d in documents)
    for document in documents:
        messages = [i.message for i in await _issues(seeded, document) if i.code == "MAPPING_WARNING"]
        assert "в файле есть другая форма BALANCE за тот же период" in messages
        issue = next(i for i in await _issues(seeded, document) if i.code == "MAPPING_WARNING")
        assert (issue.kind.value, issue.severity.value, issue.source.value) == ("warning", "info", "core")


async def test_1x_second_form_other_period(seeded: AsyncSession) -> None:
    _, documents = await _map_dict(
        seeded, [_form("BALANCE", "sheet:а"), _form("BALANCE", "sheet:б", report_date="2024-12-31")]
    )
    assert [(d.period_year, d.period_quarter) for d in documents] == [
        (2025, PeriodQuarter.Q3),
        (2024, PeriodQuarter.Q4),
    ]
    for document in documents:
        assert not any("другая форма" in i.message for i in await _issues(seeded, document))


async def test_osv_orphans_mapping_warning_issue(seeded: AsyncSession) -> None:
    sheet = _osv_sheet("60", "sheet:60")
    sheet["rows"] = [*sheet["rows"], sheet["rows"][1] | {"row_number": 40, "parent_row_number": 99}]
    file, [document] = await _map_dict(seeded, [sheet])
    issues = await _issues(seeded, document)
    [orphans] = [i for i in issues if i.code == "MAPPING_WARNING"]
    assert "без родителя" in orphans.message and "40" in orphans.message
    assert (orphans.kind.value, orphans.severity.value, orphans.source.value) == ("warning", "info", "core")
    job = await seeded.get(RecognitionJob, file.current_recognition_job_id)
    assert job is not None and any("без родителя" in w for w in job.metrics[MAPPING_WARNINGS_KEY])


# --- приём файла результата: выбор по MAJOR, нечитаемый JSON (B5) ---------------------------------------------------


def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str:
    return f"results/{file_id}/{job_id}/result.json"


async def test_result_major_3_is_fatal(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None:
    """P3-46: неизвестный MAJOR (`3.0`) — неповторяемая ошибка схемы: файл `ERROR`, без документов и повтора."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal() | {"schema_version": "3.0"})

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    job = await _job(maker, job_id)
    assert job.metrics[MAPPING_ERROR_KEY].startswith("результат не соответствует схеме")
    assert MAPPING_ATTEMPTS_KEY not in job.metrics
    assert await _documents(maker, file_id) == []
    assert await process_once(maker, storage, _settings()) is False


async def test_result_1x_still_mapped(committed_sessionmaker: async_sessionmaker[AsyncSession]) -> None:
    """MAJOR 1 (`1.2`) — маппер `1.x`: документ ОСВ создан, файл `PROCESSED`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, _, _ = await _prepare(maker, storage, _minimal() | {"schema_version": "1.2"}, owner_inn="7700000000")

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
    assert len(await _documents(maker, file_id)) == 1


@pytest.mark.parametrize("content", [b"{not json", b"", b"\xff\xfe\x00", b"[1, 2]", b"null"])
async def test_unreadable_result_json_is_fatal(
    committed_sessionmaker: async_sessionmaker[AsyncSession], content: bytes
) -> None:
    """T5.2: невалидный JSON / не объект — `MappingFatal` («результат не соответствует схеме…»): файл `ERROR`,
    без повтора, без текста исключения парсера в `mapping_error`."""
    maker, storage = committed_sessionmaker, FakeStorage()
    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
    storage.objects[_result_key(file_id, job_id)] = content

    assert await process_once(maker, storage, _settings()) is True

    assert (await _file(maker, file_id)).status is FileStatus.ERROR
    job = await _job(maker, job_id)
    assert job.metrics[MAPPING_ERROR_KEY].startswith("результат не соответствует схеме recognition-result")
    assert "Expecting" not in job.metrics[MAPPING_ERROR_KEY]
    assert MAPPING_ATTEMPTS_KEY not in job.metrics
    assert await process_once(maker, storage, _settings()) is False