arm-core/api/tests/test_recognition_inbox.py

"""Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в
`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22). Срез 2c: статусы
`stored_file` по ack / результату (spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5);
`ResultMapper` удалён — inbox не читает S3 и не создаёт документы (маппинг — отдельный цикл, D-1).

Быстрые тесты (без маркера) — разбор заголовка `Nats-Msg-Id`, слияние метрик, отказ невалидным сообщениям без
обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён.
Интеграционные — `handle_message` с фейковым Msg (`SimpleNamespace(data, headers, subject)`) на реальном PostgreSQL
(БД сессии pytest; данные фиксируются `CommittedJobs` и удаляются в teardown) и полный цикл через реальный NATS
(уникальный result-subject: общий `files.tabular_parsed` не трогаем).
"""

import asyncio
import importlib.util
import inspect
import json
import time
import uuid
from collections.abc import AsyncIterator, Iterator
from datetime import UTC, datetime, timedelta
from functools import partial
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
import pytest_asyncio
from jsonschema import Draft202012Validator, FormatChecker
from loguru import logger
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg
from sqlalchemy import JSON, func, select
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine

from src.contract import NATS_MSG_ID_HEADER, ack_msg_id, result_msg_id
from src.core.config import get_settings
from src.core.nats import NatsConnection
from src.models.enums import FileStatus, RecognitionStatus, S3BucketRole
from src.models.files import RecognitionJob, S3Object, StoredFile
from src.services import recognition_exchange as rx
from src.services.recognition_exchange import (
    FINAL_STATUSES,
    expire_deadlines,
    handle_message,
    header_msg_id,
    merge_metrics,
    publish_pending,
)
from tests.factories import CommittedJobs

API_DIR = Path(__file__).resolve().parents[1]
MESSAGE_VALIDATORS = {
    major: Draft202012Validator(
        json.loads((API_DIR / "contracts" / file_name).read_text("utf-8")), format_checker=FormatChecker()
    )
    for major, file_name in (
        ("1", "recognition-result-message.v1.schema.json"),
        ("2", "recognition-result-message.schema.json"),
    )
}

SERVICE = {"name": "tabular-parser", "version": "0.1.0"}
BUCKET = "arm-ka"
READY_TIMEOUT = 5.0
RACE_JOBS = 10
LOCK_HOLD_SECONDS = 0.5
_ERROR = {"code": "UNSUPPORTED_FORMAT", "message": "формат файла не поддерживается"}


# --- инфраструктура ---------------------------------------------------------------------------------------------


@pytest.fixture
def logs() -> Iterator[list[Any]]:
    """Записи loguru (уровень DEBUG и выше), выпущенные за время теста."""
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
    try:
        yield records
    finally:
        logger.remove(handler_id)


def _logged(records: list[Any], fragment: str, level: str | None = None) -> list[Any]:
    return [r for r in records if fragment in r["message"] and (level is None or r["level"].name == level)]


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


def envelope(
    job_id: uuid.UUID, file_id: uuid.UUID, status: str, *, version: str = "1.0", **extra: Any
) -> dict[str, Any]:
    """Конверт `recognition-result-message` (по умолчанию 1.0); для done/needs_check по умолчанию — `result_s3` и
    версия схемы результата того же MAJOR."""
    body: dict[str, Any] = {
        "schema_version": version,
        "job_id": str(job_id),
        "file_id": str(file_id),
        "status": status,
        "service": SERVICE,
    }
    if status in ("done", "needs_check"):
        body["result_s3"] = {"bucket": BUCKET, "key": _result_key(file_id, job_id)}
        body["result_schema_version"] = version
    body.update(extra)
    return body


def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER, validate: bool = True) -> Any:
    """Msg-подобный объект nats-py: `.data`, `.headers`, `.subject`. Тело по умолчанию проверяется JSON Schema."""
    if validate:
        MESSAGE_VALIDATORS[str(body["schema_version"]).split(".")[0]].validate(body)
    data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
    headers = None if msg_id is None else {header: msg_id}
    return SimpleNamespace(data=data, headers=headers, subject="files.tabular_parsed")


class SlowFileStatus:
    """Подмена `set_file_status_for_job` в inbox: запоминает вызовы (id задания, статус) и держит паузу `delay`
    ВНУТРИ транзакции сообщения (строка задания остаётся заблокированной), затем вызывает настоящую функцию."""

    def __init__(self, monkeypatch: pytest.MonkeyPatch, *, delay: float) -> None:
        self.delay = delay
        self.calls: list[tuple[uuid.UUID, FileStatus]] = []
        self._original = rx.set_file_status_for_job
        monkeypatch.setattr(rx, "set_file_status_for_job", self)

    async def __call__(self, session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool:
        self.calls.append((job.id, status))
        await asyncio.sleep(self.delay)
        return await self._original(session, job, status, **kwargs)


@pytest.fixture
def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(db_engine, expire_on_commit=False)


@pytest_asyncio.fixture
async def jobs(session_factory: async_sessionmaker[AsyncSession]) -> AsyncIterator[CommittedJobs]:
    committed = CommittedJobs(session_factory)
    try:
        yield committed
    finally:
        await committed.cleanup()


async def _row(session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> dict[str, Any]:
    """Снимок всех колонок строки `recognition_job`."""
    async with session_factory() as session:
        row = (await session.execute(select(RecognitionJob.__table__).where(RecognitionJob.id == job_id))).one()
        return dict(row._mapping)


async def _s3_rows(session_factory: async_sessionmaker[AsyncSession], key: str) -> list[dict[str, Any]]:
    async with session_factory() as session:
        rows = await session.execute(select(S3Object.__table__).where(S3Object.key == key))
        return [dict(r._mapping) for r in rows]


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


async def _file_row(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> dict[str, Any]:
    """Снимок всех колонок строки `stored_file`."""
    async with session_factory() as session:
        row = (await session.execute(select(StoredFile.__table__).where(StoredFile.id == file_id))).one()
        return dict(row._mapping)


async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID:
    return (await jobs.get(job_id)).file_id


async def _handle(session_factory: async_sessionmaker[AsyncSession], msg: Any, **kwargs: Any) -> None:
    await handle_message(msg, session_factory=session_factory, **kwargs)


# --- быстрые тесты: заголовок, метрики, разбор ------------------------------------------------------------------


@pytest.mark.parametrize(
    ("headers", "expected"),
    [
        ({"Nats-Msg-Id": "result:x:1"}, "result:x:1"),
        ({"nats-msg-id": "result:x:1"}, "result:x:1"),
        ({"NATS-MSG-ID": "ack:x:1", "other": "y"}, "ack:x:1"),
        ({"other": "y"}, None),
        ({"Nats-Msg-Id": ""}, None),
        ({}, None),
        (None, None),
    ],
)
def test_header_msg_id_is_case_insensitive(headers: dict[str, str] | None, expected: str | None):
    assert header_msg_id(headers) == expected


@pytest.mark.parametrize(
    ("existing", "incoming", "expected"),
    [
        ({"republish_count": 2}, {"duration_ms": 5}, {"republish_count": 2, "duration_ms": 5}),
        ({"republish_count": 2, "rows": 1}, {"rows": 7}, {"republish_count": 2, "rows": 7}),
        (None, {"duration_ms": 5}, {"duration_ms": 5}),  # SQL NULL / jsonb null (старые строки, сырой SQL)
        ([None, {"x": 1}], {"duration_ms": 5}, {"duration_ms": 5}),  # не объект — как пустой
        ({"republish_count": 1}, None, {"republish_count": 1}),  # метрик сервиса нет — без изменений
        (None, None, None),
    ],
)
def test_merge_metrics_keeps_republish_count(existing: Any, incoming: Any, expected: Any):
    assert merge_metrics(existing, incoming) == expected


class _UntouchableFactory:
    """session_factory, к которому нельзя обращаться: доказывает, что сообщение отвергнуто до БД."""

    def __call__(self) -> Any:
        raise AssertionError("невалидное сообщение не должно открывать сессию БД")


_JOB = uuid.UUID("11111111-1111-4111-8111-111111111111")
_FILE = uuid.UUID("22222222-2222-4222-8222-222222222222")


@pytest.mark.parametrize(
    "data",
    [
        b"{not json",
        b"\xff\xfe",
        b"[1, 2]",
        b'"text"',
        json.dumps({"schema_version": "1.0", "status": "done"}).encode(),  # нет job_id
        json.dumps(envelope(_JOB, _FILE, "done", schema_version="3.0")).encode(),  # неизвестный MAJOR
        json.dumps({k: v for k, v in envelope(_JOB, _FILE, "done").items() if k != "result_s3"}).encode(),
        json.dumps(envelope(_JOB, _FILE, "error")).encode(),  # error без поля error
        json.dumps(envelope(_JOB, _FILE, "finished")).encode(),  # статус вне enum
        json.dumps(envelope(_JOB, _FILE, "done", service={"name": "other", "version": "1"})).encode(),
        json.dumps({**envelope(_JOB, _FILE, "done"), "job_id": "not-a-uuid"}).encode(),
    ],
)
async def test_invalid_message_is_logged_and_does_not_touch_db(data: bytes, logs: list[Any]):
    msg = SimpleNamespace(data=data, headers={NATS_MSG_ID_HEADER: "result:x:1"}, subject="files.tabular_parsed")
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    rejected = _logged(logs, "отклонено", "WARNING")
    assert len(rejected) == 1
    # тело целиком в лог не попадает
    assert "result.json" not in rejected[0]["message"]


async def test_invalid_message_log_carries_job_id_when_readable(logs: list[Any]):
    body = envelope(_JOB, _FILE, "error")  # нет обязательного error
    msg = fake_msg(body, "result:x:1", validate=False)
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    (record,) = _logged(logs, "отклонено", "WARNING")
    assert str(_JOB) in record["message"]


async def test_invalid_message_job_id_is_escaped_in_log(logs: list[Any]):
    """job_id невалидного сообщения с переводом строки попадает в лог через repr: подделать строку лога нельзя (M-7)."""
    forged = "not-a-uuid\n2026-10-04 | INFO | подделанная запись\r\x1b[31m"
    body = {**envelope(_JOB, _FILE, "done"), "job_id": forged}
    msg = fake_msg(body, "result:x:1", validate=False)
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    (record,) = _logged(logs, "отклонено", "WARNING")
    message = record["message"]
    assert "\n" not in message and "\r" not in message and "\x1b" not in message
    assert repr(forged) in message


async def test_database_error_is_logged_and_not_raised(logs: list[Any]):
    """Ошибка БД (сервер недоступен) — `logger.exception`, без повторов, наружу не выходит."""
    engine = create_async_engine("postgresql+asyncpg://nobody:nothing@127.0.0.1:1/none")
    try:
        factory = async_sessionmaker(engine, expire_on_commit=False)
        await handle_message(
            fake_msg(envelope(_JOB, _FILE, "processing"), ack_msg_id(_JOB, 1)), session_factory=factory
        )
    finally:
        await engine.dispose()
    (record,) = _logged(logs, "транзакция откачена", "ERROR")
    assert record["exception"] is not None


def test_inbox_has_no_result_mapper():
    """OQ-67 (Срез 2c, D-1): точки расширения `ResultMapper` / `NoopMapper` нет, `handle_message` без `mapper`."""
    assert not hasattr(rx, "ResultMapper")
    assert not hasattr(rx, "NoopMapper")
    assert "mapper" not in inspect.signature(handle_message).parameters


def test_final_statuses():
    """R-22 (design #34): финальные для inbox — done, needs_check, error, timeout, dlq."""
    assert {s.value for s in FINAL_STATUSES} == {"done", "needs_check", "error", "timeout", "dlq"}


def test_template_nats_callback_removed():
    """R-3 (design #16): шаблонный `core/nats_callback.py` удалён и нигде в `src/` не упоминается."""
    assert importlib.util.find_spec("src.core.nats_callback") is None
    offenders = [p for p in (API_DIR / "src").rglob("*.py") if "nats_callback" in p.read_text(encoding="utf-8")]
    assert offenders == []


# --- ack: processing ---------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_ack_moves_published_job_to_processing(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Сценарий «Сервис подтвердил»."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.published, request_msg_id=f"{uuid.uuid4()}:1")
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)
    moment = datetime(2026, 10, 4, 12, 0, tzinfo=UTC)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
    )

    after = await _row(session_factory, job_id)
    assert after["status"] is RecognitionStatus.processing
    assert after["started_at"] == moment
    assert (after["service_name"], after["service_version"]) == ("tabular-parser", "0.1.0")
    changed = {k for k in after if after[k] != before[k]}
    assert changed == {"status", "started_at", "service_name", "service_version"}
    assert _logged(logs, "ack", "INFO")


@pytest.mark.integration
async def test_inbox_ack_v2_moves_job_to_processing(session_factory, jobs: CommittedJobs):
    """Ack конверта 2.0 принимается так же, как 1.0 (ядро на окне перехода держит оба MAJOR)."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.published, request_msg_id=f"{uuid.uuid4()}:1")
    file_id = await _file_id(jobs, job_id)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "processing", version="2.0"), ack_msg_id(job_id, 1))
    )

    after = await _row(session_factory, job_id)
    assert after["status"] is RecognitionStatus.processing
    assert (after["service_name"], after["service_version"]) == ("tabular-parser", "0.1.0")


@pytest.mark.integration
async def test_inbox_v1_warnings_wrapped(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    metrics = {"warnings": ["sheet=ОСВ 60 строки без родителя"]}

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "done", metrics=metrics), result_msg_id(job_id, 1))
    )

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.done
    assert job["metrics"]["warnings"] == [
        {"code": "LEGACY_WARNING", "severity": "info", "message": "sheet=ОСВ 60 строки без родителя", "locator": None}
    ]


@pytest.mark.integration
async def test_inbox_v2_warnings_kept(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    warning = {"code": "JSON_FAILED", "severity": "error", "message": "ошибочный json", "locator": "file:a.json"}
    body = envelope(job_id, file_id, "needs_check", version="2.0", metrics={"rows": 3, "warnings": [warning]})

    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.needs_check
    assert job["result_schema_version"] == "2.0"
    assert job["metrics"] == {"rows": 3, "warnings": [warning]}
    assert job["fields_total"] is None


@pytest.mark.integration
async def test_repeated_ack_changes_nothing(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.published)
    file_id = await _file_id(jobs, job_id)
    ack = fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
    await _handle(session_factory, ack)
    first = await _row(session_factory, job_id)

    await _handle(session_factory, ack, now=lambda: datetime.now(UTC) + timedelta(minutes=1))

    assert await _row(session_factory, job_id) == first


@pytest.mark.integration
@pytest.mark.parametrize(
    "status",
    [
        RecognitionStatus.queued,
        RecognitionStatus.done,
        RecognitionStatus.needs_check,
        RecognitionStatus.error,
        RecognitionStatus.timeout,
        RecognitionStatus.dlq,
    ],
)
async def test_ack_for_other_status_is_logged_and_ignored(
    session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
):
    job_id = await jobs.make_queued_job(status=status)
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == before
    (record,) = _logged(logs, "задание не изменено", "WARNING")
    assert status.value in record["message"]


# --- результат ---------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_done_result_registers_s3_object_and_updates_job(session_factory, jobs: CommittedJobs):
    """Сценарий «Успешный результат»: s3_object (results), result_msg_id, finished_at, метрики слиты."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics={"republish_count": 1})
    file_id = await _file_id(jobs, job_id)
    moment = datetime(2026, 10, 4, 12, 30, tzinfo=UTC)
    metrics = {"duration_ms": 1234, "rows": 50, "fields_total": 300, "fields_low_confidence": 4,
               "fields_unrecognized": 2, "warnings": ["лист без шапки"]}  # fmt: skip
    body = envelope(job_id, file_id, "done", metrics=metrics, result_schema_version="1.1")
    before_total = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)), now=lambda: moment)

    job = await _row(session_factory, job_id)
    (s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
    assert await _s3_total(session_factory) == before_total + 1
    assert job["status"] is RecognitionStatus.done
    assert job["result_msg_id"] == f"result:{job_id}:1"
    assert job["result_s3_object_id"] == s3["id"]
    assert job["finished_at"] == moment
    assert job["result_schema_version"] == "1.1"
    assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")
    # предупреждения 1.x ложатся структурами (LEGACY_WARNING), остальные метрики — как пришли
    legacy = [{"code": "LEGACY_WARNING", "severity": "info", "message": "лист без шапки", "locator": None}]
    assert job["metrics"] == {"republish_count": 1, **metrics, "warnings": legacy}
    assert (job["fields_total"], job["fields_low_confidence"], job["fields_unrecognized"]) == (300, 4, 2)
    assert (job["error_code"], job["error_message"]) == (None, None)
    assert s3["bucket_role"] is S3BucketRole.results
    assert (s3["bucket"], s3["created_by"], s3["created_at"]) == (BUCKET, "tabular-parser", moment)


@pytest.mark.integration
async def test_needs_check_result(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "needs_check"), result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.needs_check
    assert job["result_s3_object_id"] is not None
    assert job["metrics"] is None  # метрик не было ни у задания, ни у сервиса: колонка не трогается


@pytest.mark.integration
@pytest.mark.parametrize(
    ("stored", "expected_state"),
    [(None, (True, None)), (JSON.NULL, (False, "null"))],
    ids=["sql_null", "json_null"],
)
async def test_result_metrics_merge_over_null_metrics(
    session_factory, jobs: CommittedJobs, stored: Any, expected_state: Any
):
    """Слияние метрик в inbox (Python `merge_metrics`) работает и для SQL NULL, и для jsonb null (финальное ревью I-1);
    результат без метрик колонку не трогает: SQL NULL остаётся SQL NULL."""
    with_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
    without_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
    assert await jobs.metrics_sql_state(with_metrics) == expected_state

    file_id = await _file_id(jobs, with_metrics)
    body = envelope(with_metrics, file_id, "done", metrics={"duration_ms": 10})
    await _handle(session_factory, fake_msg(body, result_msg_id(with_metrics, 1)))
    file_id = await _file_id(jobs, without_metrics)
    await _handle(
        session_factory, fake_msg(envelope(without_metrics, file_id, "done"), result_msg_id(without_metrics, 1))
    )

    job = await _row(session_factory, with_metrics)
    assert job["status"] is RecognitionStatus.done
    assert job["metrics"] == {"duration_ms": 10}
    assert await jobs.metrics_sql_state(with_metrics) == (False, "object")
    assert (await _row(session_factory, without_metrics))["status"] is RecognitionStatus.done
    assert await jobs.metrics_sql_state(without_metrics) == expected_state


@pytest.mark.integration
async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object, ни stored_file не меняются."""
    file_id, job_id = await jobs.make_file_job(
        file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
    )
    msg = fake_msg(envelope(job_id, file_id, "done", metrics={"rows": 3}), result_msg_id(job_id, 1))
    await _handle(session_factory, msg)
    job_before = await _row(session_factory, job_id)
    file_before = await _file_row(session_factory, file_id)
    s3_before = await _s3_rows(session_factory, _result_key(file_id, job_id))
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, msg, now=lambda: datetime.now(UTC) + timedelta(hours=1))

    assert await _row(session_factory, job_id) == job_before
    assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == s3_before
    assert await _s3_total(session_factory) == total_before
    assert await _file_row(session_factory, file_id) == file_before
    assert _logged(logs, "дубликат", "DEBUG")
    assert not _logged(logs, "финальном статусе")


@pytest.mark.integration
async def test_service_error_without_result_s3(session_factory, jobs: CommittedJobs):
    """Сценарий «Ошибка сервиса»: NOT_IMPLEMENTED без result_s3."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    total_before = await _s3_total(session_factory)
    body = envelope(job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "разбор ОСВ — Срез 1"})

    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.error
    assert (job["error_code"], job["error_message"]) == ("NOT_IMPLEMENTED", "разбор ОСВ — Срез 1")
    assert job["result_s3_object_id"] is None
    assert job["result_msg_id"] == f"result:{job_id}:1"
    assert job["finished_at"] is not None
    assert job["result_schema_version"] is None
    assert await _s3_total(session_factory) == total_before


@pytest.mark.integration
@pytest.mark.parametrize(
    "status",
    [
        RecognitionStatus.timeout,
        RecognitionStatus.dlq,
        RecognitionStatus.error,
        RecognitionStatus.done,
        RecognitionStatus.needs_check,
    ],
)
async def test_result_for_final_status_is_logged_and_ignored(
    session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
):
    """Сценарий «Результат пришёл после таймаута» (и прочие финальные статусы, R-22): задание не меняется."""
    other = f"result:{uuid.uuid4()}:9" if status in (RecognitionStatus.done, RecognitionStatus.needs_check) else None
    job_id = await jobs.make_queued_job(status=status, result_msg_id=other)
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == before
    assert await _s3_total(session_factory) == total_before
    (record,) = _logged(logs, "финальном статусе", "WARNING")
    assert status.value in record["message"]


@pytest.mark.integration
async def test_unknown_job_is_logged_and_creates_nothing(session_factory, logs: list[Any]):
    job_id, file_id = uuid.uuid4(), uuid.uuid4()
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    assert await _s3_total(session_factory) == total_before
    assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == []
    (record,) = _logged(logs, "неизвестного задания", "WARNING")
    assert str(job_id) in record["message"]


@pytest.mark.integration
async def test_file_id_mismatch_is_logged_and_ignored(session_factory, jobs: CommittedJobs, logs: list[Any]):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    before = await _row(session_factory, job_id)
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(envelope(job_id, uuid.uuid4(), "done"), result_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == before
    assert await _s3_total(session_factory) == total_before
    assert _logged(logs, "file_id", "WARNING")


@pytest.mark.integration
@pytest.mark.parametrize("status", [RecognitionStatus.published, RecognitionStatus.queued])
async def test_result_without_ack_is_accepted(session_factory, jobs: CommittedJobs, status: RecognitionStatus):
    """R-22: потерянный ack не блокирует результат; `started_at` остаётся пустым (начало обработки неизвестно)."""
    job_id = await jobs.make_queued_job(status=status)
    file_id = await _file_id(jobs, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.done
    assert job["started_at"] is None
    assert job["service_name"] == "tabular-parser"


@pytest.mark.integration
async def test_msg_id_header_case_insensitive(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1), header="nats-msg-id")
    )

    assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"


@pytest.mark.integration
async def test_missing_msg_id_header_uses_derived_id(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Нет `Nats-Msg-Id` — предупреждение и производный `result:<job_id>:<attempt>`; повтор — дубликат."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, attempt=2)
    file_id = await _file_id(jobs, job_id)
    msg = fake_msg(envelope(job_id, file_id, "done"), None)

    await _handle(session_factory, msg)
    job = await _row(session_factory, job_id)
    await _handle(session_factory, msg)

    assert job["status"] is RecognitionStatus.done
    assert job["result_msg_id"] == f"result:{job_id}:2"
    assert await _row(session_factory, job_id) == job
    assert len(_logged(logs, "без заголовка Nats-Msg-Id", "WARNING")) == 2  # noqa: PLR2004 - оба сообщения


@pytest.mark.integration
async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs):
    """Ключ результата уже зарегистрирован — строка переиспользуется, дубль (bucket, key) не вставляется."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    async with session_factory() as session, session.begin():
        existing = S3Object(
            bucket_role=S3BucketRole.results,
            bucket=BUCKET,
            key=_result_key(file_id, job_id),
            created_by="tabular-parser",
            created_at=datetime.now(UTC),
        )
        session.add(existing)
        await session.flush()
        existing_id = existing.id
    jobs.track_s3_object(existing_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    assert (await _row(session_factory, job_id))["result_s3_object_id"] == existing_id
    assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1


# --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) -------------


@pytest.mark.integration
async def test_ack_moves_uploaded_file_to_processing(session_factory, jobs: CommittedJobs):
    """Сценарий «Сервис подтвердил»: файл текущего задания — `PROCESSING`, `updated_at` — момент ack."""
    file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.published)
    moment = datetime(2026, 10, 5, 9, 0, tzinfo=UTC)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
    )

    file = await _file_row(session_factory, file_id)
    assert file["status"] is FileStatus.PROCESSING
    assert file["updated_at"] == moment
    assert file["current_recognition_job_id"] == job_id
    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing


@pytest.mark.integration
@pytest.mark.parametrize(
    "file_status", [FileStatus.ERROR, FileStatus.PROCESSED, FileStatus.SUPERSEDED, FileStatus.PROCESSING]
)
async def test_ack_moves_file_only_from_uploaded(session_factory, jobs: CommittedJobs, file_status: FileStatus):
    """R-5: ack ставит `PROCESSING` только из `UPLOADED` — запоздавший ack не воскрешает `ERROR` и не трогает
    `PROCESSED`/`SUPERSEDED`; задание при этом подтверждается как обычно."""
    file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.published)
    before = await _file_row(session_factory, file_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))

    assert await _file_row(session_factory, file_id) == before
    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing


@pytest.mark.integration
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED])
async def test_error_result_moves_file_to_error(session_factory, jobs: CommittedJobs, file_status: FileStatus):
    """Сценарий «Парсер вернул ошибку»: файл — `ERROR`, причина — в задании (`error_code`)."""
    file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing)
    moment = datetime(2026, 10, 5, 9, 30, tzinfo=UTC)

    await _handle(
        session_factory,
        fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1)),
        now=lambda: moment,
    )

    file = await _file_row(session_factory, file_id)
    assert (file["status"], file["updated_at"]) == (FileStatus.ERROR, moment)
    assert (await _row(session_factory, job_id))["error_code"] == "UNSUPPORTED_FORMAT"


@pytest.mark.integration
@pytest.mark.parametrize("status", ["done", "needs_check"])
async def test_success_result_leaves_processing_file_for_mapping(session_factory, jobs: CommittedJobs, status: str):
    """Сценарий «Успешный результат»: файл остаётся `PROCESSING` до маппинга; строка файла не меняется вовсе."""
    file_id, job_id = await jobs.make_file_job(
        file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
    )
    before = await _file_row(session_factory, file_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1)))

    assert await _file_row(session_factory, file_id) == before
    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)


@pytest.mark.integration
@pytest.mark.parametrize("status", ["done", "needs_check"])
@pytest.mark.parametrize("job_status", [RecognitionStatus.published, RecognitionStatus.queued])
async def test_success_result_without_ack_moves_uploaded_file_to_processing(
    session_factory, jobs: CommittedJobs, status: str, job_status: RecognitionStatus
):
    """R-4: ack потерялся (core NATS без JetStream) — результат `done`/`needs_check` переводит файл `UPLOADED →
    PROCESSING`, иначе файл с готовым результатом не попал бы в цикл маппинга."""
    file_id, job_id = await jobs.make_file_job(job_status=job_status)
    moment = datetime(2026, 10, 5, 10, 0, tzinfo=UTC)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1)), now=lambda: moment
    )

    file = await _file_row(session_factory, file_id)
    assert (file["status"], file["updated_at"]) == (FileStatus.PROCESSING, moment)
    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)


@pytest.mark.integration
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
@pytest.mark.parametrize("status", ["done", "error"])
async def test_result_does_not_touch_processed_or_superseded_file(
    session_factory, jobs: CommittedJobs, file_status: FileStatus, status: str
):
    """`PROCESSED` и `SUPERSEDED` — итог маппинга / автозамены: результат обмена их не перезаписывает."""
    file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing)
    before = await _file_row(session_factory, file_id)
    extra = {"error": _ERROR} if status == "error" else {}

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, status, **extra), result_msg_id(job_id, 1)))

    assert await _file_row(session_factory, file_id) == before
    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)


@pytest.mark.integration
@pytest.mark.parametrize(
    ("status", "msg_id_fn", "expected_job"),
    [
        ("processing", ack_msg_id, RecognitionStatus.processing),
        ("done", result_msg_id, RecognitionStatus.done),
        ("needs_check", result_msg_id, RecognitionStatus.needs_check),
        ("error", result_msg_id, RecognitionStatus.error),
    ],
)
async def test_message_of_non_current_job_does_not_touch_file(  # noqa: PLR0913 - фикстуры и параметры
    session_factory, jobs: CommittedJobs, status: str, msg_id_fn: Any, expected_job: RecognitionStatus
):
    """Review Focus 1: у файла два задания, текущее — новое; ack/результат старого задания применяется к старому
    заданию, но файл (статус, `updated_at`, `current_recognition_job_id`) не трогает."""
    file_id, old_job = await jobs.make_file_job(job_status=RecognitionStatus.published)
    new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)
    before = await _file_row(session_factory, file_id)
    assert before["current_recognition_job_id"] == new_job
    extra = {"error": _ERROR} if status == "error" else {}

    await _handle(session_factory, fake_msg(envelope(old_job, file_id, status, **extra), msg_id_fn(old_job, 1)))

    assert (await _row(session_factory, old_job))["status"] is expected_job
    assert await _file_row(session_factory, file_id) == before
    assert (await _row(session_factory, new_job))["status"] is RecognitionStatus.queued


@pytest.mark.integration
async def test_file_status_rolls_back_with_message_transaction(
    session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch
):
    """Статус файла меняется в транзакции сообщения: ошибка после обновления задания откатывает и задание, и файл."""
    file_id, job_id = await jobs.make_file_job(
        file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
    )
    job_before = await _row(session_factory, job_id)
    file_before = await _file_row(session_factory, file_id)
    original = rx.set_file_status_for_job

    async def failing(session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool:
        await original(session, job, status, **kwargs)
        raise RuntimeError("подставная ошибка после обновления файла")

    monkeypatch.setattr(rx, "set_file_status_for_job", failing)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == job_before
    assert await _file_row(session_factory, file_id) == file_before
    (record,) = _logged(logs, "транзакция откачена", "ERROR")
    assert record["exception"] is not None


# --- гонки --------------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_concurrent_same_message_applied_once(
    session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch
):
    file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.processing)
    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
    slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS)  # первый держит строку, второй ждёт его commit

    await asyncio.gather(*[_handle(session_factory, msg) for _ in range(2)])

    assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1
    assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"
    assert slow.calls == [(job_id, FileStatus.PROCESSING)]  # применён один раз
    assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
    assert len(_logged(logs, "дубликат", "DEBUG")) == 1
    assert not _logged(logs, "неизвестного задания")


class _SlowNats:
    """Подставной NATS: публикация «висит» `delay` секунд, пока публикатор держит блокировку строки задания."""

    def __init__(self, delay: float) -> None:
        self.delay = delay
        self.entered = asyncio.Event()

    @property
    def is_connected(self) -> bool:
        return True

    async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
        self.entered.set()
        await asyncio.sleep(self.delay)


@pytest.mark.integration
async def test_ack_waits_for_publisher_lock(session_factory, jobs: CommittedJobs):
    """Ack пришёл, пока публикатор держит строку (до его commit): обработчик ждёт блокировку и применяет ack после."""
    job_id = await jobs.make_queued_job()
    file_id = await _file_id(jobs, job_id)
    nats = _SlowNats(LOCK_HOLD_SECONDS)
    publisher = asyncio.create_task(
        publish_pending(session_factory, nats, ack_timeout=timedelta(seconds=30), max_republish=3, batch_size=1)
    )
    await asyncio.wait_for(nats.entered.wait(), 5)
    started = time.monotonic()

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))

    waited = time.monotonic() - started
    report = await publisher
    job = await _row(session_factory, job_id)
    assert report.published == 1
    assert waited >= LOCK_HOLD_SECONDS / 2  # ждал блокировку, а не увидел queued
    assert job["status"] is RecognitionStatus.processing
    assert job["published_at"] is not None
    assert job["started_at"] is not None


@pytest.mark.integration
async def test_result_vs_watchdog_is_consistent(session_factory, jobs: CommittedJobs):
    """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object, файл
    `PROCESSING`), либо timeout (без результата, файл `ERROR`), но не смесь. Порядок решает блокировка строки задания:
    кто первый, тот и финальный (и задание, и файл меняются в порядке «задание → файл», без взаимоблокировок)."""
    past = datetime.now(UTC) - timedelta(minutes=1)
    pairs = [
        await jobs.make_file_job(
            file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
        )
        for _ in range(RACE_JOBS)
    ]
    job_ids = [job_id for _, job_id in pairs]
    messages = [fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) for file_id, job_id in pairs]

    await asyncio.gather(expire_deadlines(session_factory), *[_handle(session_factory, m) for m in messages])

    for job_id in job_ids:
        job = await _row(session_factory, job_id)
        s3 = await _s3_rows(session_factory, _result_key(job["file_id"], job_id))
        file = await _file_row(session_factory, job["file_id"])
        if job["status"] is RecognitionStatus.done:
            assert job["error_code"] is None
            assert job["result_msg_id"] == f"result:{job_id}:1"
            assert [r["id"] for r in s3] == [job["result_s3_object_id"]]
            assert file["status"] is FileStatus.PROCESSING
        else:
            assert file["status"] is FileStatus.ERROR
            assert job["status"] is RecognitionStatus.timeout
            assert job["error_code"] == "TIMEOUT"
            assert job["result_msg_id"] is None
            assert job["result_s3_object_id"] is None
            assert s3 == []


@pytest.mark.integration
async def test_watchdog_waits_for_result_transaction(
    session_factory, jobs: CommittedJobs, monkeypatch: pytest.MonkeyPatch
):
    """Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой
    версии строки (`done`) и задание не трогает; файл остаётся `PROCESSING` (не `ERROR`)."""
    past = datetime.now(UTC) - timedelta(minutes=1)
    file_id, job_id = await jobs.make_file_job(
        file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
    )
    slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS)
    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
    handler = asyncio.create_task(_handle(session_factory, msg))

    async def locked() -> None:  # обработчик внутри транзакции, строка заблокирована
        while not slow.calls:
            await asyncio.sleep(0.01)

    await asyncio.wait_for(locked(), 5)

    expired = await expire_deadlines(session_factory)
    await handler

    assert expired == 0
    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.done
    assert job["error_code"] is None
    assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING


@pytest.mark.integration
async def test_watchdog_after_result_and_result_after_watchdog(session_factory, jobs: CommittedJobs):
    past = datetime.now(UTC) - timedelta(minutes=1)
    first = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
    second = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)

    await _handle(
        session_factory, fake_msg(envelope(first, await _file_id(jobs, first), "done"), result_msg_id(first, 1))
    )
    await expire_deadlines(session_factory)
    await _handle(
        session_factory, fake_msg(envelope(second, await _file_id(jobs, second), "done"), result_msg_id(second, 1))
    )

    assert (await _row(session_factory, first))["status"] is RecognitionStatus.done
    assert (await _row(session_factory, second))["status"] is RecognitionStatus.timeout


# --- e2e: реальный NATS -------------------------------------------------------------------------------------------


async def _wait_until(predicate: Any, timeout: float = 10.0) -> None:
    deadline = time.monotonic() + timeout
    while not await predicate():
        if time.monotonic() > deadline:
            raise AssertionError("условие не выполнено за отведённое время")
        await asyncio.sleep(0.05)


async def _wait_status(
    session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
) -> dict[str, Any]:
    async def reached() -> bool:
        return (await _row(session_factory, job_id))["status"] is status

    await _wait_until(reached)
    return await _row(session_factory, job_id)


class _FakeService:
    """«Сервис распознавания» поверх голого nats-py: принимает запросы своего задания, публикует конверты."""

    def __init__(self, job_id: uuid.UUID, result_subject: str) -> None:
        self.job_id = job_id
        self.result_subject = result_subject
        self.requests: list[Msg] = []
        self._nc = NATS()

    async def start(self, url: str) -> None:
        await self._nc.connect(url, connect_timeout=2, allow_reconnect=False)

        async def collect(msg: Msg) -> None:
            if json.loads(msg.data).get("job_id") == str(self.job_id):
                self.requests.append(msg)

        await self._nc.subscribe("files.tabular_uploaded", cb=collect)
        await self._nc.flush()

    async def send(self, body: dict[str, Any], msg_id: str) -> None:
        MESSAGE_VALIDATORS[str(body["schema_version"]).split(".")[0]].validate(body)
        await self._nc.publish(self.result_subject, json.dumps(body).encode(), headers={NATS_MSG_ID_HEADER: msg_id})
        await self._nc.flush()

    async def stop(self) -> None:
        await self._nc.close()


@pytest.mark.integration
async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """queued → published (outbox) → processing (ack) → done + s3_object (результат) → повтор без изменений;
    файл: UPLOADED → PROCESSING (ack) → PROCESSING (результат, ждёт цикл маппинга)."""
    settings = get_settings().OPENBAO.NATS
    result_subject = f"test.result.{uuid.uuid4().hex}"
    handler = partial(handle_message, session_factory=session_factory)

    async def subscribe(conn: NatsConnection) -> None:
        await conn.subscribe_results(handler)

    core = NatsConnection(settings, on_connected=subscribe, result_subject=result_subject)
    file_id, job_id = await jobs.make_file_job()
    service = _FakeService(job_id, result_subject)
    await core.start()
    try:
        if not await core.wait_ready(timeout=READY_TIMEOUT):
            pytest.fail(f"NATS недоступен ({settings.NATS_URL}, .env.test)", pytrace=False)
        await service.start(settings.NATS_URL)

        async def published() -> bool:
            await publish_pending(session_factory, core, ack_timeout=timedelta(seconds=30), max_republish=3)
            return bool(service.requests)

        await _wait_until(published)
        await _wait_status(session_factory, job_id, RecognitionStatus.published)
        assert service.requests[0].headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}

        await service.send(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
        job = await _wait_status(session_factory, job_id, RecognitionStatus.processing)
        assert job["started_at"] is not None
        assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
        assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")

        result = envelope(job_id, file_id, "done", metrics={"duration_ms": 10})
        await service.send(result, result_msg_id(job_id, 1))
        job = await _wait_status(session_factory, job_id, RecognitionStatus.done)
        (s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
        assert (job["result_s3_object_id"], job["result_msg_id"]) == (s3["id"], f"result:{job_id}:1")
        assert job["finished_at"] is not None
        assert s3["bucket_role"] is S3BucketRole.results
        assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING

        await service.send(result, result_msg_id(job_id, 1))

        async def duplicate_seen() -> bool:
            return bool(_logged(logs, "дубликат", "DEBUG"))

        await _wait_until(duplicate_seen)
        assert await _row(session_factory, job_id) == job
        assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == [s3]
    finally:
        await service.stop()
        await core.stop()