"""Сквозной сценарий scan-gateway против compose-инфраструктуры (задача 6.1, R-7, R-10; Срез 3a).

Настоящий lifespan приложения (`create_app` → S3, NATS, воркер, `OcrNormalizer` на копии `arm_statements`) и
эмулятор ИС Сканирование (`src.scan_emulator.run` в потоке, `SCAN_EMULATOR_DEFAULT` пуст) работают в процессе
теста; NATS 127.0.0.1:4222 и MinIO 127.0.0.1:9000 — из `api/.env.test` (`make up`), горячая папка — tmp-каталог.
Входящий файл — `<job_id>.pdf` (расширение `original_name` в нижнем регистре, пишется прямо в `Input`), пару фикстур
эмулятор находит по sha256 содержимого: в его каталоге фикстур лежат PDF и ответы (реальные `aep_*` из
`fixtures/ocr` и синтетический `ok_doc`).

Сценарии (реальные фикстуры `fixtures/ocr`: OQ-76/77/78; эмулятор кладёт в `Output/<job_id>/` весь набор ответа):
1. `aep_5.PDF` (ответ-ошибка `INVALID_DOCUMENT`) → ack `processing`, затем `INVALID_DOCUMENT` с сообщением ИС и
   `raw_s3_key`; повтор запроса во время ожидания ответа ИС и после результата — ещё один ack, второго файла нет;
2. `aep_3.pdf` (набор из двух файлов: `<job_id>.json` — успех, `<job_id>_ofr.json` — `INVALID_DOCUMENT`) → ack, затем
   `needs_check` с документами (баланс и ОФР) в файле результата и `metrics.warnings`, в том числе про `_ofr.json`;
   `raw/{job_id}.json` — оба файла (склейка, D-4);
3. `aep_6.pdf` (один успешный файл) → `needs_check`, `result_s3` с двумя документами (`BALANCE`, `INCOME`);
4. `ok_doc.pdf` (синтетический успешный ответ в форме реальных) → результат с `result_s3` (имя строки из OCR);
5. `unknown.pdf` (содержимого нет в фикстурах) → ack, затем `INVALID_DOCUMENT` эмулятора;
6. `aep_6.pdf` при `SCAN_CLEANUP=true` → после результата `Input/` и `Output/<job_id>/` пусты.
Каждый конверт проверяется схемой `recognition-result-message`, `service.name` и `Nats-Msg-Id`; файл результата
из MinIO — схемой `recognition-result`; склеенный сырой ответ в MinIO содержит ровно те файлы, что лежат в каталоге
ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора
— по settle (`SETTLE_SECONDS`, 0.2 с).

Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Тест слушает ответы на своём
subject'е прогона; конверты отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-<uuid>/…`,
`raw/<job_id>.json`, `results/…`) и удаляются.
Подписка шлюза идёт в queue group `scan-gateway`. Тест работает на своих subject'ах прогона (`TEST_SUBJECTS`), поэтому
экземпляры стенда на умолчательных `files.*` его запросы не получают. Недоступная инфраструктура — падение теста с
понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`).
"""

import asyncio
import hashlib
import json
import shutil
import socket
import threading
import uuid
from collections.abc import AsyncIterator
from contextlib import AsyncExitStack, suppress
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import httpx
import pytest
from fastapi import FastAPI
from jsonschema import Draft202012Validator, FormatChecker
from miniopy_async.error import S3Error
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg

from src.app import create_app
from src.contract import CONTRACT_VERSION, NATS_MSG_ID_HEADER, request_msg_id
from src.core.clients import JSON_CONTENT_TYPE, S3Storage
from src.core.config import Settings, get_settings
from src.core.nats import used_subjects
from src.scan_emulator import EmulatorSettings, run

pytestmark = [
    pytest.mark.integration,
    pytest.mark.usefixtures("log_lines"),
    # miniopy-async отдаёт aiohttp тело объекта байтами; на теле > 1 МиБ (реальный скан aep_5.PDF) aiohttp пишет
    # ResourceWarning «Sending a large body directly…». Предупреждение библиотеки, по умолчанию Python его не
    # показывает; здесь отключено точечно, чтобы прогон с `-W error` проверял остальные предупреждения
    pytest.mark.filterwarnings("ignore:Sending a large body directly with raw bytes:ResourceWarning"),
]

API_DIR = Path(__file__).resolve().parents[2]
CONTRACTS_DIR = API_DIR / "contracts"
# `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как в unit-тестах
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"

GOLDEN_JOB_ID = "3f1c2b7a-0000-4000-8000-000000000001"  # условный job_id эталонов (`tests/golden_update.py`)
# Уникальные subject'ы прогона (OPENBAO__NATS__NATS_*_SUBJECT): шлюз теста и слушатель работают на них, контейнеры
# стенда на умолчательных `files.*` запросы теста не получают и не отвечают на них
RUN = uuid.uuid4().hex[:8]
TEST_SUBJECTS = {
    "image_request": f"itest.{RUN}.files.img_uploaded",
    "result": f"itest.{RUN}.files.tabular_parsed",
}
CONNECT_WAIT = 10.0  # секунд на подключение к NATS (тест и фоновая задача приложения)
SCENARIO_WAIT = 20.0  # секунд на ожидание конвертов одного сценария
QUIET_PERIOD = 1.0  # секунд тишины после повторного запроса: второго результата быть не должно
EMULATOR_DELAY = 0.5  # секунд «распознавания»: окно, в котором повторный запрос приходит во время ожидания ИС
POLL_SECONDS = 0.1  # SCAN_POLL_SECONDS шлюза
SETTLE_SECONDS = 0.2  # SCAN_SETTLE_SECONDS шлюза: набор ответа готов, когда не менялся столько секунд
TIMEOUT_SECONDS = 15  # SCAN_TIMEOUT_SECONDS: зависший эмулятор даёт TIMEOUT раньше, чем истечёт SCENARIO_WAIT

MESSAGE_VALIDATOR = Draft202012Validator(
    json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
    format_checker=FormatChecker(),
)
RESULT_VALIDATOR = Draft202012Validator(
    json.loads((CONTRACTS_DIR / "recognition-result.schema.json").read_text(encoding="utf-8")),
    format_checker=FormatChecker(),
)
REQUEST_EXAMPLE: dict[str, Any] = json.loads(
    (CONTRACTS_DIR / "examples" / "request-image.json").read_text(encoding="utf-8")
)

# Синтетический успешный ответ ИС (R-10) в форме реальных ответов (`aep_6.json`): `data_ocr` — объект с полями
# `label/organization/inn/units/date` (`form_name` + `confidence`) и `table_structure[{label, rows}]`. Все
# `confidence` = 100, шапка есть, даты колонок и отчёта разобраны, но форма — одна строка «БАЛАНС»
OK_DOC_PDF = b"%PDF-1.4\n% scan-gateway itest ok_doc.pdf\n\x00\x01\x02"
UNKNOWN_PDF = b"%PDF-1.4\n% scan-gateway itest unknown.pdf\n\x00\x01\x02"
OK_DOC_ORGANIZATION = "ООО «Русский бисквит»"
SUCCESS_DATA_OCR: dict[str, Any] = {
    "label": {"form_name": "Бухгалтерская отчетность", "confidence": 100.0},
    "organization": {"form_name": OK_DOC_ORGANIZATION, "confidence": 100.0},
    "inn": {"form_name": "7701234567", "confidence": 100.0},
    "units": {"form_name": "тыс. руб.", "confidence": 100.0},
    "date": {"form_name": "30.09.2025", "confidence": 100.0},
    "table_structure": [
        {
            "label": {"form_name": "Бухгалтерский баланс", "confidence": 100.0},
            "rows": [
                {
                    "row_index": 0,
                    "cells": [
                        {"column_index": 0, "value": "Наименование показателя", "confidence": 100.0},
                        {"column_index": 1, "value": "Код", "confidence": 100.0},
                        {"column_index": 2, "value": "На 30 сентября 2025 г.", "confidence": 100.0},
                        {"column_index": 3, "value": "На 31 декабря 2024 г.", "confidence": 100.0},
                    ],
                },
                {
                    "row_index": 1,
                    "cells": [
                        {"column_index": 0, "value": "БАЛАНС", "confidence": 100.0},
                        {"column_index": 1, "value": "1600", "confidence": 100.0},
                        {"column_index": 2, "value": "1 234 567", "confidence": 100.0},
                        {"column_index": 3, "value": "(12 345)", "confidence": 100.0},
                    ],
                },
            ],
        }
    ],
}


def _require_port(host: str, port: int, what: str) -> None:
    try:
        with socket.create_connection((host, port), timeout=2):
            return
    except OSError as exc:
        pytest.fail(f"{what} недоступен на {host}:{port} ({exc!r}): поднимите инфраструктуру `make up`")


def _real_fixture(stem: str, suffix: str) -> Path:
    """Файл реальной фикстуры без учёта регистра расширения (`aep_5.PDF`); нет репозитория fixtures — skip."""
    if not REAL_FIXTURES.is_dir():
        pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
    for path in sorted(REAL_FIXTURES.iterdir()):
        if path.stem == stem and path.suffix.lower() == suffix:
            return path
    pytest.fail(f"в {REAL_FIXTURES} нет {stem}{suffix} (без учёта регистра расширения)")


@dataclass
class Envelope:
    msg_id: str
    body: dict[str, Any]


@dataclass
class Roundtrip:
    """Запущенный шлюз (lifespan приложения), эмулятор и слушатель `files.tabular_parsed`."""

    settings: Settings
    app: FastAPI
    root: Path
    s3: S3Storage
    publisher: NATS
    received: list[Msg] = field(default_factory=list)
    jobs: set[str] = field(default_factory=set)
    keys: list[str] = field(default_factory=list)

    @property
    def input_dir(self) -> Path:
        return self.root / self.settings.SCAN_INPUT_DIR

    def answer_dir(self, job_id: str) -> Path:
        return self.root / self.settings.SCAN_OUTPUT_DIR / job_id

    def answer_names(self, job_id: str) -> list[str]:
        """Имена всех `*.json` каталога ответа по возрастанию (каталог шлюз не удаляет)."""
        return sorted(p.name for p in self.answer_dir(job_id).glob("*.json"))

    # --- запрос ---------------------------------------------------------------------------------------------------
    async def request(self, original_name: str, data: bytes) -> dict[str, Any]:
        """Кладёт исходник в MinIO под уникальным ключом и строит запрос по `examples/request-image.json`."""
        job_id, file_id = str(uuid.uuid4()), str(uuid.uuid4())
        key = f"uploads/itest-{uuid.uuid4()}/{file_id}/{job_id}/{original_name}"
        self.jobs.add(job_id)
        self.keys += [key, f"raw/{job_id}.json", f"results/{file_id}/{job_id}/result.json"]
        await self.s3.put_bytes(key, data, content_type="application/pdf")
        deadline = datetime.now(UTC) + timedelta(seconds=60)
        return {
            **REQUEST_EXAMPLE,
            "job_id": job_id,
            "file_id": file_id,
            "original_name": original_name,
            "s3": {"bucket": self.s3.bucket, "key": key},
            "reply_deadline": deadline.isoformat().replace("+00:00", "Z"),
            "attempt": 1,
        }

    async def publish(self, body: dict[str, Any]) -> None:
        """Публикация как в ядре: `files.img_uploaded`, `Nats-Msg-Id = <job_id>:<attempt>`."""
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        headers = {NATS_MSG_ID_HEADER: request_msg_id(body["job_id"], body["attempt"])}
        await self.publisher.publish(TEST_SUBJECTS["image_request"], payload, headers=headers)
        await self.publisher.flush()

    # --- конверты -------------------------------------------------------------------------------------------------
    def envelopes(self, job_id: str) -> list[Envelope]:
        """Конверты задания в порядке получения; каждый проверен схемой и `service.name`."""
        out = []
        for msg in self.received:
            body = json.loads(msg.data)
            if body.get("job_id") != job_id:
                continue
            MESSAGE_VALIDATOR.validate(body)
            assert body["service"]["name"] == "scan-gateway", body
            assert body["schema_version"] == CONTRACT_VERSION == "2.0", body  # ack и итог — 2.0 (P5-13)
            assert msg.headers is not None, body
            out.append(Envelope(msg.headers[NATS_MSG_ID_HEADER], body))
        return out

    def acks(self, job_id: str) -> list[str]:
        return [e.msg_id for e in self.envelopes(job_id) if e.body["status"] == "processing"]

    def results(self, job_id: str) -> list[Envelope]:
        return [e for e in self.envelopes(job_id) if e.body["status"] != "processing"]

    async def wait_for(self, what: str, predicate: Any, timeout: float = SCENARIO_WAIT) -> None:
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        while not predicate():
            if loop.time() > deadline:
                raise AssertionError(f"за {timeout} с не дождались: {what}")
            await asyncio.sleep(0.01)

    async def health(self) -> dict[str, Any]:
        transport = httpx.ASGITransport(app=self.app)
        async with httpx.AsyncClient(transport=transport, base_url="http://scan-gateway") as client:
            response = await client.get("/healthcheck")
        assert response.status_code == 200
        body: dict[str, Any] = response.json()
        return body

    # --- проверки -------------------------------------------------------------------------------------------------
    async def raw(self, job_id: str) -> dict[str, Any]:
        raw: dict[str, Any] = json.loads(await self.s3.get_bytes(f"raw/{job_id}.json"))
        return raw

    def inbox(self, job_id: str) -> list[str]:
        return sorted(p.name for p in self.input_dir.iterdir() if p.name.startswith(job_id))

    async def run(self, original_name: str, data: bytes, *, cleanup: bool = False) -> tuple[dict[str, Any], Envelope]:
        """Запрос → ack → итоговый конверт; общие проверки сквозного сценария. Возвращает запрос и итог."""
        body = await self.request(original_name, data)
        job_id = body["job_id"]
        await self.publish(body)
        await self.wait_for(f"итоговый конверт {job_id}", lambda: bool(self.results(job_id)))
        envelopes = self.envelopes(job_id)
        assert [e.msg_id for e in envelopes] == [f"ack:{job_id}:1", f"result:{job_id}:1"]  # ack раньше итога
        assert envelopes[0].body["status"] == "processing"
        assert envelopes[0].body["file_id"] == body["file_id"]
        await self.check_files(body, envelopes[1], data, cleanup=cleanup)
        return body, envelopes[1]

    async def check_files(self, body: dict[str, Any], result: Envelope, data: bytes, *, cleanup: bool = False) -> None:
        """Сырой ответ в MinIO — склейка ответа эмулятора (D-4); входящий файл `<job_id>.pdf` = объект S3.

        Без очистки (`SCAN_CLEANUP=false`) каталог ответа и входящий файл (с суффиксом `.done` эмулятора) остаются; при
        очистке (`SCAN_CLEANUP=true`) во входящем каталоге пусто, каталога `Output/<job_id>/` нет.
        """
        job_id = body["job_id"]
        assert result.body["file_id"] == body["file_id"]
        assert result.body["raw_s3_key"] == f"raw/{job_id}.json"
        raw = await self.raw(job_id)
        assert raw["job_id"] == job_id
        assert raw["files"], "в склейке нет ни одного файла"
        for entry in raw["files"]:
            assert entry["parse_error"] is None
            assert entry["response"]["file_uid"] == job_id
        stat = await self.s3.client.stat_object(self.s3.bucket, f"raw/{job_id}.json")
        assert stat.content_type == JSON_CONTENT_TYPE
        if cleanup:
            assert list(self.input_dir.iterdir()) == []
            assert not self.answer_dir(job_id).exists()
            return
        names = self.answer_names(job_id)
        assert names, "эмулятор не положил ни одного json"
        # склейка (D-4): все json каталога ответа, по возрастанию имени, тела совпадают с файлами
        assert [entry["name"] for entry in raw["files"]] == names
        for entry in raw["files"]:
            assert entry["response"] == json.loads((self.answer_dir(job_id) / entry["name"]).read_bytes())
        # входящий файл — `<job_id>.<ext>` в нижнем регистре; эмулятор переименовал его в `.done`
        name = f"{job_id}.pdf.done"
        assert self.inbox(job_id) == [name]
        assert (self.input_dir / name).read_bytes() == await self.s3.get_bytes(body["s3"]["key"]) == data

    async def check_error(self, body: dict[str, Any], result: Envelope) -> None:
        """Итог `error`: файла результата в MinIO нет (он пишется только для done/needs_check)."""
        assert result.body["status"] == "error"
        assert "result_s3" not in result.body
        with pytest.raises(S3Error):
            await self.s3.client.stat_object(self.s3.bucket, f"results/{body['file_id']}/{body['job_id']}/result.json")

    async def check_needs_check(self, body: dict[str, Any], result: Envelope) -> dict[str, Any]:
        """Итог `needs_check` с `result_s3`; файл результата из MinIO проходит схему `recognition-result`.

        Возвращает файл результата (документы лежат только в нём: конверт несёт ссылку `result_s3`).
        """
        job_id, file_id = body["job_id"], body["file_id"]
        assert result.body["status"] == "needs_check"
        assert "error" not in result.body
        key = f"results/{file_id}/{job_id}/result.json"
        assert result.body["result_s3"] == {"bucket": self.s3.bucket, "key": key}
        stat = await self.s3.client.stat_object(self.s3.bucket, key)
        assert stat.content_type == JSON_CONTENT_TYPE
        payload: dict[str, Any] = json.loads(await self.s3.get_bytes(key))
        RESULT_VALIDATOR.validate(payload)
        assert payload["schema_version"] == CONTRACT_VERSION
        assert (payload["job_id"], payload["file_id"], payload["status"]) == (job_id, file_id, "needs_check")
        assert payload["service"]["name"] == "scan-gateway"
        assert payload["raw_s3_key"] == f"raw/{job_id}.json"
        # итог в конверте и в файле результата — одни и те же метрики
        assert payload["metrics"] == result.body["metrics"]
        return payload


def _write_fixtures(fixtures: Path) -> None:
    """Каталог фикстур эмулятора: пары PDF+JSON по содержимому (sha256). Реальные — копии `aep_3` (`aep_3.json` +
    `aep_3_ofr.json`), `aep_5`, `aep_6` из `fixtures/ocr`; синтетическая — `ok_doc.pdf` + `ok_doc.json` (R-10), её байты
    сценарий 4 отправляет в S3. Без репозитория fixtures — только синтетическая пара."""
    fixtures.mkdir()
    template: dict[str, Any] = {"file_uid": "aep_5"}
    if REAL_FIXTURES.is_dir():
        for path in sorted(REAL_FIXTURES.iterdir()):
            if path.stem in ("aep_3", "aep_3_ofr", "aep_5", "aep_6") and path.suffix.lower() in (".json", ".pdf"):
                shutil.copy(path, fixtures / path.name)
        template = json.loads((fixtures / "aep_5.json").read_text(encoding="utf-8"))
    # успех — в форме реальных ответов ИС: `error = {"code": "", "message": ""}`, а не `null` (OQ-77)
    success = {**template, "file_uid": "ok_doc", "error": {"code": "", "message": ""}, "data_ocr": SUCCESS_DATA_OCR}
    (fixtures / "ok_doc.pdf").write_bytes(OK_DOC_PDF)
    (fixtures / "ok_doc.json").write_text(json.dumps(success, ensure_ascii=False, indent=2), encoding="utf-8")


def _settings(monkeypatch: pytest.MonkeyPatch, root: Path, *, cleanup: bool) -> Settings:
    """Настройки шлюза из `.env.test` и окружения (как в проде) с tmp-корнем; инфраструктура недоступна — fail."""
    monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local")
    monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(root))
    monkeypatch.setenv("SCAN_POLL_SECONDS", str(POLL_SECONDS))
    monkeypatch.setenv("SCAN_SETTLE_SECONDS", str(SETTLE_SECONDS))
    monkeypatch.setenv("SCAN_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS))
    monkeypatch.setenv("SCAN_CLEANUP", "true" if cleanup else "false")
    monkeypatch.setenv("NATS_ENABLED", "true")
    monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", TEST_SUBJECTS["image_request"])
    monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", TEST_SUBJECTS["result"])
    get_settings.cache_clear()
    settings = get_settings()
    assert used_subjects(settings.OPENBAO.NATS) == TEST_SUBJECTS
    parsed = urlparse(settings.OPENBAO.NATS.NATS_URL)
    _require_port(parsed.hostname or "127.0.0.1", parsed.port or 4222, "NATS")
    host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":")
    _require_port(host, int(port or 9000), "MinIO")
    return settings


async def _start_gateway(stack: AsyncExitStack, app: FastAPI) -> None:
    """Lifespan приложения; возврат — когда шлюз подписан на subject запроса."""
    await stack.enter_async_context(app.router.lifespan_context(app))
    [connect_task] = [t for t in asyncio.all_tasks() if t.get_name() == "nats-connect"]
    # задача завершается после SUB с flush: дальше подписка шлюза действует на сервере
    await asyncio.wait_for(asyncio.shield(connect_task), CONNECT_WAIT)
    assert app.state.nats.is_connected, "шлюз не подписался на запросы (см. лог ERROR)"


def _start_emulator(stack: AsyncExitStack, settings: Settings, root: Path, fixtures: Path) -> None:
    """Эмулятор ИС (`src.scan_emulator.run`) в потоке; при закрытии стека — штатная остановка и ожидание потока."""
    stop = threading.Event()
    emulator_settings = EmulatorSettings(
        hotfolder_path=root,
        input_dir=settings.SCAN_INPUT_DIR,
        output_dir=settings.SCAN_OUTPUT_DIR,
        tmp_dir="Tmp",  # служебный каталог эмулятора; у шлюза его нет (change hotfolder-direct-write)
        fixtures_dir=fixtures,
        delay=EMULATOR_DELAY,
        poll_interval=0.05,
        default="",  # `SCAN_EMULATOR_DEFAULT` пуст: промах по хешу даёт ответ-ошибку, а не набор по умолчанию
    )
    emulator = asyncio.create_task(asyncio.to_thread(run, emulator_settings, stop), name="scan-emulator")

    async def stop_emulator() -> None:
        stop.set()
        await emulator

    stack.push_async_callback(stop_emulator)


async def _roundtrip(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str], *, cleanup: bool
) -> AsyncIterator[Roundtrip]:
    """Запущенные шлюз и эмулятор; `cleanup` — значение `SCAN_CLEANUP` шлюза."""
    root = tmp_path / "hotfolder"
    root.mkdir()
    _write_fixtures(tmp_path / "fixtures")
    settings = _settings(monkeypatch, root, cleanup=cleanup)
    assert settings.SCAN_CLEANUP is cleanup
    nats_url = settings.OPENBAO.NATS.NATS_URL

    async with AsyncExitStack() as stack:
        s3 = S3Storage.from_settings(settings.OPENBAO.MINIO)
        stack.push_async_callback(s3.close)
        client = NATS()
        try:
            await asyncio.wait_for(client.connect(nats_url, max_reconnect_attempts=0), CONNECT_WAIT)
        except Exception as exc:
            pytest.fail(f"нет подключения к NATS {nats_url} ({exc!r}): `make up`")
        stack.push_async_callback(client.drain)
        rt = Roundtrip(settings=settings, app=create_app(settings), root=root, s3=s3, publisher=client)

        async def remove_objects() -> None:
            for key in rt.keys:
                with suppress(S3Error):
                    await s3.client.remove_object(s3.bucket, key)

        stack.push_async_callback(remove_objects)

        async def on_result(msg: Msg) -> None:
            with suppress(ValueError):  # чужие сообщения на общем subject могут быть чем угодно
                body = json.loads(msg.data)
                if isinstance(body, dict) and body.get("job_id") in rt.jobs:
                    rt.received.append(msg)

        # core NATS без JetStream не хранит сообщения: слушатель подписан до первой публикации запроса
        await client.subscribe(TEST_SUBJECTS["result"], cb=on_result)
        await client.flush()
        await _start_gateway(stack, rt.app)
        _start_emulator(stack, settings, root, tmp_path / "fixtures")
        yield rt

    # lifespan остановлен: заданий и фоновых задач приложения не осталось, ошибок в логе нет
    assert rt.app.state.worker.active_jobs == 0
    leaked = [t.get_name() for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
    assert leaked == [], f"незавершённые задачи после остановки: {leaked}"
    errors = [line for line in log_lines if line.startswith(("ERROR|", "CRITICAL|"))]
    assert errors == [], errors


@pytest.fixture
async def roundtrip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str]) -> AsyncIterator[Roundtrip]:
    """Шлюз с `SCAN_CLEANUP=false` (по умолчанию)."""
    async for rt in _roundtrip(tmp_path, monkeypatch, log_lines, cleanup=False):
        yield rt


@pytest.fixture
async def roundtrip_cleanup(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str]
) -> AsyncIterator[Roundtrip]:
    """Шлюз с `SCAN_CLEANUP=true`."""
    async for rt in _roundtrip(tmp_path, monkeypatch, log_lines, cleanup=True):
        yield rt


def _golden(name: str, job_id: str) -> dict[str, Any] | None:
    """Эталон v2 `fixtures/ocr/expected_normalize/<name>.json` с подставленным `job_id`; нет эталона — `None`.

    Эталон строится на условном `job_id` (`tests/golden_update.py`): он встречается в `locator` и в предупреждениях.
    """
    path = REAL_FIXTURES / "expected_normalize" / f"{name}.json"
    if not path.is_file():
        return None
    expected: dict[str, Any] = json.loads(path.read_text(encoding="utf-8").replace(GOLDEN_JOB_ID, job_id))
    return expected


@pytest.mark.fixtures
async def test_real_fixture_aep_5_passes_is_error_through_and_repeat_gets_ack(roundtrip: Roundtrip) -> None:
    pdf = _real_fixture("aep_5", ".pdf")  # реальное имя `aep_5.PDF`: расширение в верхнем регистре
    data = pdf.read_bytes()
    expected_error = json.loads(_real_fixture("aep_5", ".json").read_text(encoding="utf-8"))["error"]
    health = await roundtrip.health()
    assert (health["nats"], health["hotfolder"]) == ("ok", "ok"), health

    body = await roundtrip.request(pdf.name, data)
    job_id = body["job_id"]
    await roundtrip.publish(body)
    # расширение входящего файла — в нижнем регистре, имя файла = job_id (раскладка FileWorker)
    inbox_file = roundtrip.input_dir / f"{job_id}.pdf"
    await roundtrip.wait_for(f"файл {inbox_file.name} во входящем каталоге", inbox_file.exists)
    # повтор того же запроса во время ожидания ответа ИС (эмулятор «распознаёт» EMULATOR_DELAY с)
    await roundtrip.publish(body)
    await roundtrip.wait_for(f"итоговый конверт {job_id}", lambda: bool(roundtrip.results(job_id)))
    await asyncio.sleep(QUIET_PERIOD)  # второй (лишний) результат или ack успел бы прийти

    envelopes = roundtrip.envelopes(job_id)
    assert [e.msg_id for e in envelopes] == [f"ack:{job_id}:1", f"ack:{job_id}:1", f"result:{job_id}:1"]
    [result] = roundtrip.results(job_id)
    assert result.body["error"] == {"code": "INVALID_DOCUMENT", "message": expected_error["message"]}
    assert result.body["error"]["message"] == "Не удалось определить тип документа."
    await roundtrip.check_files(body, result, data)
    await roundtrip.check_error(body, result)
    raw = await roundtrip.raw(job_id)
    assert [entry["response"]["error"] for entry in raw["files"]] == [expected_error]
    # все файлы набора ошибочные (Р-32/Р-33): итог — ошибка первого файла, а warnings перечисляют `JSON_FAILED`
    # каждого ошибочного файла структурой
    assert result.body["metrics"]["warnings"] == [
        {
            "code": "JSON_FAILED",
            "severity": "error",
            "message": f"INVALID_DOCUMENT: {expected_error['message']}",
            "locator": f"file:{job_id}.json",
        }
    ]

    # повтор ядра после результата (attempt 2): ещё один ack, без второго файла и второго результата
    await roundtrip.publish({**body, "attempt": 2})
    await roundtrip.wait_for(f"ack:{job_id}:2", lambda: f"ack:{job_id}:2" in roundtrip.acks(job_id))
    await asyncio.sleep(QUIET_PERIOD)
    assert roundtrip.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:1", f"ack:{job_id}:2"]
    assert [e.msg_id for e in roundtrip.results(job_id)] == [f"result:{job_id}:1"]
    assert roundtrip.inbox(job_id) == [f"{job_id}.pdf.done"]


@pytest.mark.fixtures
async def test_real_fixture_aep_3_two_files_merged_needs_check_with_documents(roundtrip: Roundtrip) -> None:
    """Набор из двух файлов: успешный `<job_id>.json` и `<job_id>_ofr.json` с `INVALID_DOCUMENT` (OQ-76, OQ-78)."""
    data = _real_fixture("aep_3", ".pdf").read_bytes()
    ofr_error = json.loads(_real_fixture("aep_3_ofr", ".json").read_text(encoding="utf-8"))["error"]
    assert ofr_error["code"] == "INVALID_DOCUMENT"

    body, result = await roundtrip.run("aep_3.pdf", data)
    job_id = body["job_id"]
    payload = await roundtrip.check_needs_check(body, result)
    # документы собраны из успешного файла: баланс и ОФР, по одному на таблицу `table_structure`
    assert [d["document_type"] for d in payload["documents"]] == ["BALANCE", "INCOME"]
    assert [d["locator"] for d in payload["documents"]] == [
        f"file:{job_id}.json;table:0",
        f"file:{job_id}.json;table:1",
    ]
    assert payload["metrics"]["rows"] == sum(len(d["lines"]) for d in payload["documents"])
    # ошибочный файл набора — `JSON_FAILED` структурой (`error`, `locator` файла); предупреждения документов лежат
    # в самих документах (§11.9), а не в `metrics.warnings`
    warnings = result.body["metrics"]["warnings"]
    assert warnings == [
        {
            "code": "JSON_FAILED",
            "severity": "error",
            "message": f"INVALID_DOCUMENT: {ofr_error['message']}",
            "locator": f"file:{job_id}_ofr.json",
        }
    ]
    assert payload["metrics"]["warnings"] == warnings
    expected = _golden("aep_3", job_id)
    if expected is not None:  # эталон нормализатора (`make golden-update`) совпадает со сквозным результатом
        assert payload["documents"] == expected["documents"]
        assert warnings == expected["metrics"]["warnings"]

    raw = await roundtrip.raw(job_id)
    assert [entry["name"] for entry in raw["files"]] == [f"{job_id}.json", f"{job_id}_ofr.json"]
    first, second = raw["files"]
    assert first["response"]["error"] == {"code": "", "message": ""}  # успех — не `null`
    assert [t["label"]["form_name"] for t in first["response"]["data_ocr"]["table_structure"]] == [
        "Бухгалтерский баланс",
        "Отчет о финансовых результатах",
    ]
    assert second["response"]["error"] == ofr_error
    assert (first["parse_error"], second["parse_error"]) == (None, None)


@pytest.mark.fixtures
async def test_real_fixture_aep_6_needs_check_with_two_documents(roundtrip: Roundtrip) -> None:
    data = _real_fixture("aep_6", ".pdf").read_bytes()
    body, result = await roundtrip.run("aep_6.pdf", data)
    job_id = body["job_id"]
    payload = await roundtrip.check_needs_check(body, result)
    assert [d["document_type"] for d in payload["documents"]] == ["BALANCE", "INCOME"]
    # ошибочных файлов в наборе нет: предупреждений файла нет, документы равны эталону
    assert result.body["metrics"].get("warnings", []) == []
    expected = _golden("aep_6", job_id)
    if expected is not None:
        assert payload["documents"] == expected["documents"]
        assert result.body["metrics"].get("warnings", []) == expected["metrics"].get("warnings", [])
    raw = await roundtrip.raw(job_id)
    [entry] = raw["files"]
    assert entry["name"] == f"{job_id}.json"
    assert entry["response"]["error"] == {"code": "", "message": ""}


async def test_synthetic_success_answer_gives_needs_check(roundtrip: Roundtrip) -> None:
    """`ok_doc.pdf` лежит в фикстурах эмулятора теми же байтами, что отправляются в S3 (пара по sha256, R-10)."""
    body, result = await roundtrip.run("ok_doc.pdf", OK_DOC_PDF)
    job_id = body["job_id"]
    payload = await roundtrip.check_needs_check(body, result)
    [document] = payload["documents"]
    assert document["document_type"] == "BALANCE"
    assert document["locator"] == f"file:{job_id}.json;table:0"
    assert document["header"]["organization_name"] == OK_DOC_ORGANIZATION
    assert document["header"]["inn"] == "7701234567"
    assert document["header"]["report_date"] == "2025-09-30"
    assert [c["period_date"] for c in document["columns"]] == ["2025-09-30", "2024-12-31"]
    [line] = document["lines"]
    assert (line["line_code"], line["line_name"]) == ("1600", "БАЛАНС")  # имя из OCR (`line_name`)
    assert [v["amount"] for v in line["values"]] == ["1234567.00", "-12345.00"]  # деньги — строки numeric(18,2)
    # предупреждения «редакция не заполнена» больше нет (редакции — в пакете); предупреждений файла нет
    assert result.body["metrics"].get("warnings", []) == []
    assert not any("редакция" in json.dumps(w, ensure_ascii=False) for w in document.get("warnings", []))
    raw = await roundtrip.raw(job_id)
    [entry] = raw["files"]
    assert entry["response"]["error"] == {"code": "", "message": ""}
    assert entry["response"]["data_ocr"] == SUCCESS_DATA_OCR


async def test_unknown_content_gets_emulator_invalid_document(roundtrip: Roundtrip) -> None:
    """Содержимого нет среди фикстур, `SCAN_EMULATOR_DEFAULT` пуст: ответ-ошибка эмулятора (D-3)."""
    body, result = await roundtrip.run("unknown.pdf", UNKNOWN_PDF)
    job_id = body["job_id"]
    digest = hashlib.sha256(UNKNOWN_PDF).hexdigest()
    error = {"code": "INVALID_DOCUMENT", "message": f"{digest[:12]}: нет пары по содержимому в фикстурах"}
    await roundtrip.check_error(body, result)
    assert result.body["error"] == error
    raw = await roundtrip.raw(job_id)
    [entry] = raw["files"]
    assert entry["name"] == f"{job_id}.json"
    assert entry["parse_error"] is None
    # ответ-заглушка эмулятора в форме реальных ответов ИС (R-28): `data_ocr` — объект без таблиц
    assert entry["response"]["file_uid"] == job_id
    assert entry["response"]["error"] == error
    assert entry["response"]["data_ocr"]["table_structure"] == []


@pytest.mark.fixtures
async def test_cleanup_enabled_leaves_input_and_answer_empty(roundtrip_cleanup: Roundtrip) -> None:
    """`SCAN_CLEANUP=true`: после результата во входящем каталоге пусто, каталога `Output/<job_id>/` нет (D-2)."""
    data = _real_fixture("aep_6", ".pdf").read_bytes()
    body, result = await roundtrip_cleanup.run("aep_6.pdf", data, cleanup=True)
    job_id = body["job_id"]
    await roundtrip_cleanup.check_needs_check(body, result)
    # очистка идёт после сохранения сырого ответа: он и файл результата на месте, горячая папка чиста
    raw = await roundtrip_cleanup.raw(job_id)
    assert [entry["name"] for entry in raw["files"]] == [f"{job_id}.json"]
    assert list(roundtrip_cleanup.input_dir.iterdir()) == []
    assert not roundtrip_cleanup.answer_dir(job_id).exists()