CLAUDE.md

@@ -13,7 +13,7 @@
   `pytest`, `ruff check . && ruff format --check .`, `mypy src`. Из `web/`: `npm run start|build`, `npx biome check src`,
   `npx tsc --noEmit`. Бутстрап — `setup/setup.sh` (pip/npm через Nexus банка).
 - **Раскладка `api/src/`**: `api/endpoints` → `services` → `repositories` → `models`; DTO — `schemas/`; инфраструктура —
-  `core/` (config, clients, nats_callback, logger, exception_handlers, auth); `db/` — engine и сессия. Наши добавления:
+  `core/` (config, clients, nats, logger, exception_handlers, auth); `db/` — engine и сессия. Наши добавления:
   `domain/` (чистые функции по §§ правил), `api/scripts/` (`check_dbml_vs_orm.py`, seed).
 - **DI** — ручные фабрики `Depends` в `api/dependencies/`; сессия из `get_async_session`; `storage`/`nats` — из `request.app.state`.
 - **Настройки** — pydantic-settings, `api/.env`, вложенность `__`, инфраструктура под `OPENBAO__*`; `DEBUG=true`

README.md

@@ -296,7 +296,7 @@ api/
     app.py             FastAPI-приложение: lifespan (S3, NATS, SMB), /healthcheck, раздача SPA
     api/endpoints/     эндпоинты (me.py, dev_auth.py); router.py собирает /api/v1
     api/dependencies/  зависимости FastAPI (current_user, проверка прав)
-    core/              config, auth (JWT), clients (S3), nats (подключение, publish, подписка), nats_callback,
+    core/              config, auth (JWT), clients (S3), nats (подключение, publish, подписка),
                        logger, exception_handlers
     db/                движок и сессия (схема dkb)
     models/            ORM-модели, 1:1 с DBML

api/src/app.py

@@ -6,12 +6,15 @@ group `arm-core`), SMB (горячие папки ИС Сканирование)
 healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA.
 Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
 состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и
-вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS.
+вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения
+`recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий
+приложения и `NoopMapper` (маппинг результата в доменные таблицы — Срез 2).
 """
 
 import os
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
+from functools import partial
 from http import HTTPStatus
 
 import smbclient
@@ -28,9 +31,9 @@ from src.api.router import API_PREFIX, router
 from src.core.clients import S3Storage
 from src.core.config import get_settings
 from src.core.exception_handlers import register_exception_handlers
-from src.core.nats import NatsConnection
+from src.core.nats import MessageHandler, NatsConnection
 from src.db.db import async_session_maker
-from src.services.recognition_exchange import ExchangeTasks, handle_message
+from src.services.recognition_exchange import ExchangeTasks, NoopMapper, handle_message
 
 settings = get_settings()
 
@@ -91,9 +94,14 @@ async def _connect_s3() -> S3Storage | None:
     return S3Storage(client=s3_client, bucket_name=settings.OPENBAO.MINIO.MINIO_BUCKET)
 
 
+def _result_handler() -> MessageHandler:
+    """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения и `NoopMapper` (Срез 0)."""
+    return partial(handle_message, session_factory=async_session_maker, mapper=NoopMapper())
+
+
 async def _subscribe_results(nats: NatsConnection) -> None:
     """Подписки после первого подключения к NATS: `recognition.result` в queue group `arm-core`."""
-    await nats.subscribe_results(handle_message)
+    await nats.subscribe_results(_result_handler())
 
 
 def _create_nats() -> NatsConnection | None:

api/src/core/nats_callback.py

@@ -1,19 +0,0 @@
-"""Обработчики сообщений NATS.
-
-Каркас: доменный обработчик удалён. Consumer `recognition.result` (дедуп по recognition_job.result_msg_id,
-маппинг JSON из S3 в document/osv_entry/financial_statement_*) реализуется в Срезе 0–2 по
-docs/context/03-architecture.md. Для референса — reference-old-arm/api/src/core/nats_callback.py.
-"""
-
-from loguru import logger
-from nats.aio.client import Client as NATS
-
-from src.core.clients import S3Storage
-
-
-def build_result_handler(nc: NATS, storage: S3Storage):
-    async def handler(msg):
-        logger.info("NATS message received. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data))
-        await msg.ack()
-
-    return handler

api/src/services/recognition_exchange.py

@@ -22,29 +22,54 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро
 
 Обмен не пишет `audit_event` (рулинг R-7, design #20).
 
-Inbox (`handle_message`) — пока заглушка: подписка в `src/core/nats.py` уже вызывает её; полноценная обработка
-(ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23).
+Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `recognition.result`
+(`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO:
+`SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit,
+применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`.
+- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`; повторный ack для `processing` —
+  тихо игнорируется; ack для другого статуса — warning, задание не меняется.
+- результат (`done | needs_check | error`): дубликат определяется ТОЛЬКО по `recognition_job.result_msg_id ==
+  Nats-Msg-Id` (at-least-once) — без изменений; задание в финальном статусе (`FINAL_STATUSES`) — warning, без изменений;
+  иначе (`queued | published | processing`: потерянный ack не блокирует результат) — регистрация файла результата в
+  `s3_object` (`bucket_role = results`), обновление задания, `ResultMapper.map_result` (Срез 0 — `NoopMapper`; Срез 2
+  заменит маппингом результата в `document` / `osv_entry` / `financial_statement_*`) и commit.
+- Гонка с вахтёром (`expire_deadlines` — один `UPDATE` по статусу) решается блокировкой строки: кто первым взял
+  строку, тот и задаёт финальный статус; второй видит финальный статус и ничего не меняет.
+- Невалидное сообщение, неизвестный `job_id`, ошибка БД или mapper'а — лог без повторов; исключения наружу не выходят.
+  Задание без применённого результата дойдёт до `timeout` (осознанный компромисс D-6).
 """
 
 import asyncio
 import json
 import uuid
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import Any, Protocol
 
 from loguru import logger
-from nats.aio.msg import Msg
 from pydantic import ValidationError
 from sqlalchemy import CursorResult, and_, case, func, literal, or_, select, update
 from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from src.contract import CONTRACT_VERSION, RecognitionRequest, S3Ref, request_msg_id, request_subject
+from src.contract import (
+    CONTRACT_VERSION,
+    NATS_MSG_ID_HEADER,
+    MessageStatus,
+    Metrics,
+    RecognitionRequest,
+    RecognitionResultMessage,
+    S3Ref,
+    ack_msg_id,
+    request_msg_id,
+    request_subject,
+    result_msg_id,
+)
 from src.core.config import Settings
 from src.core.nats import NatsPublishError
-from src.models.enums import RecognitionPipeline, RecognitionStatus
+from src.models.enums import RecognitionPipeline, RecognitionStatus, S3BucketRole
 from src.models.files import RecognitionJob, S3Object, StoredFile
 
 WATCHDOG_PERIOD_SECONDS = 30  # период вахтёра сроков (spec «Таймаут ожидания результата»)
@@ -433,6 +458,329 @@ class ExchangeTasks:
                 )
 
 
-async def handle_message(msg: Msg) -> None:
-    """Заглушка: логирует subject и размер полученного сообщения, ничего не меняет."""
-    logger.info("Получено сообщение NATS. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data))
+# --- inbox: recognition.result ----------------------------------------------------------------------------------
+
+# Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется
+FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset(
+    {
+        RecognitionStatus.done,
+        RecognitionStatus.needs_check,
+        RecognitionStatus.error,
+        RecognitionStatus.timeout,
+        RecognitionStatus.dlq,
+    }
+)
+# Итог сервиса → статус задания
+RESULT_STATUSES: Mapping[MessageStatus, RecognitionStatus] = {
+    MessageStatus.DONE: RecognitionStatus.done,
+    MessageStatus.NEEDS_CHECK: RecognitionStatus.needs_check,
+    MessageStatus.ERROR: RecognitionStatus.error,
+}
+# Счётчики значений из метрик сервиса → одноимённые колонки задания (DBML v5.7)
+FIELD_COUNTERS = ("fields_total", "fields_low_confidence", "fields_unrecognized")
+LOG_ID_MAX_LENGTH = 64  # job_id невалидного сообщения в логе — не длиннее (тело в лог не пишется)
+
+
+class InboundMessage(Protocol):
+    """То, что inbox читает из сообщения nats-py (`nats.aio.msg.Msg`)."""
+
+    @property
+    def data(self) -> bytes: ...
+
+    @property
+    def headers(self) -> Mapping[str, str] | None: ...
+
+    @property
+    def subject(self) -> str: ...
+
+
+class ResultMapper(Protocol):
+    """Точка расширения inbox (design D-6): маппинг принятого результата в доменные таблицы.
+
+    Вызывается в транзакции обработки сообщения, после обновления задания (`job` уже содержит статус и ссылку на
+    файл результата и может быть уточнено mapper'ом). Исключение откатывает всю транзакцию: задание не меняется и
+    дойдёт до `timeout`. Сетевой IO внутри — на совести реализации: транзакция держит блокировку строки задания.
+    Срез 0 — `NoopMapper`; Срез 2 заменит чтением файла результата и записью `document` / `osv_entry` / …
+    """
+
+    async def map_result(
+        self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage
+    ) -> None: ...
+
+
+class NoopMapper:
+    """Mapper Среза 0: результат только регистрируется (`s3_object`, задание), доменные таблицы не заполняются."""
+
+    async def map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None:
+        logger.debug("Маппинг результата не выполняется (NoopMapper, Срез 0). job_id={job_id}", job_id=message.job_id)
+
+
+class _MapperError(Exception):
+    """Исключение `ResultMapper.map_result` — отличаем от ошибок БД в логе."""
+
+
+def header_msg_id(headers: Mapping[str, str] | None) -> str | None:
+    """Значение заголовка `Nats-Msg-Id` без учёта регистра имени; нет заголовка или пустое значение — `None`."""
+    if not headers:
+        return None
+    target = NATS_MSG_ID_HEADER.lower()
+    for name, value in headers.items():
+        if name.lower() == target and value:
+            return value
+    return None
+
+
+def merge_metrics(existing: Any, incoming: Mapping[str, Any] | None) -> Any:
+    """Метрики сервиса поверх метрик задания: ключи ядра (`republish_count`) сохраняются, совпадающие — от сервиса.
+
+    `incoming is None` — метрики задания как есть. Не объект (SQL NULL, JSON `null` — так ORM пишет `None`) — как `{}`.
+    """
+    if incoming is None:
+        return existing
+    base = dict(existing) if isinstance(existing, dict) else {}
+    return {**base, **incoming}
+
+
+def _parse_message(msg: InboundMessage) -> RecognitionResultMessage | None:
+    """Конверт `recognition-result-message` v1.0 или `None` (warning в лог без тела сообщения)."""
+    try:
+        data = json.loads(msg.data)
+    except ValueError:  # JSONDecodeError и UnicodeDecodeError
+        data = None
+    if not isinstance(data, dict):
+        logger.warning(
+            "Сообщение {subject} отклонено: тело не JSON-объект. error_category=invalid_message, size={size}",
+            subject=msg.subject,
+            size=len(msg.data),
+        )
+        return None
+    try:
+        return RecognitionResultMessage.model_validate(data)
+    except ValidationError as exc:
+        raw_job_id = data.get("job_id")
+        job_id = raw_job_id[:LOG_ID_MAX_LENGTH] if isinstance(raw_job_id, str) else None
+        errors = "; ".join(
+            f"{'.'.join(str(part) for part in error['loc']) or '<root>'}: {error['msg']}"
+            for error in exc.errors(include_url=False, include_input=False, include_context=False)
+        )
+        logger.warning(
+            "Сообщение {subject} отклонено: не соответствует контракту recognition-result-message. "
+            "error_category=invalid_message, job_id={job_id}, errors={errors}",
+            subject=msg.subject,
+            job_id=job_id,
+            errors=errors,
+        )
+        return None
+
+
+async def _lock_job(session: AsyncSession, job_id: uuid.UUID) -> RecognitionJob | None:
+    """Строка задания под `FOR UPDATE` — ЖДЁТ блокировку (не SKIP LOCKED / NOWAIT): ack или результат, пришедший, пока
+    публикатор или вахтёр держат строку, применяется после их commit и видит их итог."""
+    stmt = (
+        select(RecognitionJob)
+        .where(RecognitionJob.id == job_id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    return (await session.execute(stmt)).scalar_one_or_none()
+
+
+def _apply_ack(job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime) -> None:
+    """Ack (`status = processing`): `published` → `processing`; повтор для `processing` — тихо; иначе — warning."""
+    if job.status is RecognitionStatus.published:
+        job.status = RecognitionStatus.processing
+        job.started_at = now
+        job.service_name = message.service.name.value
+        job.service_version = message.service.version
+        logger.info(
+            "Сервис подтвердил получение запроса (ack): задание в processing. job_id={job_id}, service={service}, "
+            "msg_id={msg_id}",
+            job_id=job.id,
+            service=message.service.name.value,
+            msg_id=msg_id,
+        )
+    elif job.status is RecognitionStatus.processing:
+        logger.debug("Повторный ack для задания в processing, игнорируется. job_id={job_id}", job_id=job.id)
+    else:
+        logger.warning(
+            "Ack для задания не в статусе published: задание не изменено. error_category=unexpected_ack, "
+            "job_id={job_id}, status={status}, msg_id={msg_id}",
+            job_id=job.id,
+            status=job.status.value,
+            msg_id=msg_id,
+        )
+
+
+async def _register_result_object(session: AsyncSession, message: RecognitionResultMessage, now: datetime) -> uuid.UUID:
+    """Файл результата в реестре `s3_object` (`bucket_role = results`, `created_by = service.name`).
+
+    Уникальность `(bucket, key)`: строка с этим ключом уже есть — переиспользуется (дубль не вставляется, не падаем).
+    """
+    assert message.result_s3 is not None
+    bucket, key = message.result_s3.bucket, message.result_s3.key
+    stmt = (
+        pg_insert(S3Object)
+        .values(
+            bucket_role=S3BucketRole.results,
+            bucket=bucket,
+            key=key,
+            created_by=message.service.name.value,
+            created_at=now,
+        )
+        .on_conflict_do_nothing(index_elements=[S3Object.bucket, S3Object.key])
+        .returning(S3Object.id)
+    )
+    inserted = (await session.execute(stmt)).scalar_one_or_none()
+    if inserted is not None:
+        return inserted
+    existing = (
+        await session.execute(select(S3Object.id).where(S3Object.bucket == bucket, S3Object.key == key))
+    ).scalar_one()
+    logger.warning(
+        "Файл результата уже зарегистрирован в s3_object, строка переиспользована. job_id={job_id}, bucket={bucket}, "
+        "key={key}",
+        job_id=message.job_id,
+        bucket=bucket,
+        key=key,
+    )
+    return existing
+
+
+async def _apply_result(
+    session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime
+) -> bool:
+    """Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе —
+    регистрация файла результата и обновление задания (`True`, дальше — mapper и commit)."""
+    if job.result_msg_id == msg_id:
+        logger.debug(
+            "Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, "
+            "msg_id={msg_id}",
+            job_id=job.id,
+            msg_id=msg_id,
+        )
+        return False
+    if job.status in FINAL_STATUSES:
+        logger.warning(
+            "Результат для задания в финальном статусе не применён. error_category=late_result, job_id={job_id}, "
+            "status={status}, result_status={result_status}, msg_id={msg_id}",
+            job_id=job.id,
+            status=job.status.value,
+            result_status=message.status.value,
+            msg_id=msg_id,
+        )
+        return False
+
+    result_s3_object_id = await _register_result_object(session, message, now) if message.result_s3 else None
+    job.status = RESULT_STATUSES[message.status]
+    job.service_name = message.service.name.value
+    job.service_version = message.service.version
+    job.finished_at = now
+    if message.metrics is not None:
+        job.metrics = merge_metrics(job.metrics, _metrics_json(message.metrics))
+        for name in FIELD_COUNTERS:
+            value = getattr(message.metrics, name)
+            if value is not None:
+                setattr(job, name, value)
+    job.error_code = message.error.code.value if message.error is not None else None
+    job.error_message = message.error.message if message.error is not None else None
+    job.result_schema_version = message.result_schema_version
+    job.result_s3_object_id = result_s3_object_id
+    job.result_msg_id = msg_id
+    return True
+
+
+async def _map_result(
+    session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, mapper: ResultMapper
+) -> None:
+    """`mapper.map_result` в транзакции сообщения; его исключение — `_MapperError` (откат всей транзакции)."""
+    try:
+        await mapper.map_result(session, job, message)
+    except Exception as exc:
+        raise _MapperError(f"ResultMapper.map_result: {type(exc).__name__}") from exc
+    logger.info(
+        "Результат распознавания принят. job_id={job_id}, status={status}, service={service}, msg_id={msg_id}, "
+        "error_code={error_code}",
+        job_id=job.id,
+        status=job.status.value,
+        service=message.service.name.value,
+        msg_id=job.result_msg_id,
+        error_code=job.error_code,
+    )
+
+
+def _metrics_json(metrics: Metrics) -> dict[str, Any]:
+    return metrics.model_dump(mode="json", exclude_none=True)
+
+
+async def _apply_message(
+    session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, mapper: ResultMapper, clock: Clock
+) -> None:
+    job = await _lock_job(session, message.job_id)
+    if job is None:
+        logger.warning(
+            "Сообщение recognition.result для неизвестного задания, не применено. error_category=unknown_job, "
+            "job_id={job_id}, status={status}",
+            job_id=message.job_id,
+            status=message.status.value,
+        )
+        return
+    if job.file_id != message.file_id:
+        logger.warning(
+            "Сообщение recognition.result: file_id не совпадает с заданием, не применено. "
+            "error_category=file_mismatch, job_id={job_id}, file_id={file_id}, message_file_id={message_file_id}",
+            job_id=job.id,
+            file_id=job.file_id,
+            message_file_id=message.file_id,
+        )
+        return
+    is_ack = message.status is MessageStatus.PROCESSING
+    msg_id = header_id
+    if msg_id is None:
+        msg_id = ack_msg_id(job.id, job.attempt) if is_ack else result_msg_id(job.id, job.attempt)
+        logger.warning(
+            "Сообщение recognition.result без заголовка Nats-Msg-Id: использован производный id. "
+            "error_category=missing_msg_id, job_id={job_id}, msg_id={msg_id}",
+            job_id=job.id,
+            msg_id=msg_id,
+        )
+    now = clock()
+    if is_ack:
+        _apply_ack(job, message, msg_id, now)
+    elif await _apply_result(session, job, message, msg_id, now):
+        await _map_result(session, job, message, mapper)
+
+
+async def handle_message(
+    msg: InboundMessage,
+    *,
+    session_factory: async_sessionmaker[AsyncSession],
+    mapper: ResultMapper | None = None,
+    now: Clock | None = None,
+) -> None:
+    """Обработчик `recognition.result` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает
+    результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля.
+
+    Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД или mapper'а — откат транзакции,
+    `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7).
+    """
+    message = _parse_message(msg)
+    if message is None:
+        return
+    try:
+        async with session_factory() as session, session.begin():
+            await _apply_message(session, message, header_msg_id(msg.headers), mapper or NoopMapper(), now or _utcnow)
+    except _MapperError:
+        logger.exception(
+            "Ошибка маппинга результата распознавания, транзакция откачена; задание дойдёт до timeout. "
+            "error_category=mapper_error, job_id={job_id}",
+            job_id=message.job_id,
+        )
+    except Exception as exc:
+        logger.exception(
+            "Ошибка обработки сообщения {subject}, транзакция откачена (без повторов). error_category=inbox_error, "
+            "job_id={job_id}, error_type={error_type}, error={error}",
+            subject=msg.subject,
+            job_id=message.job_id,
+            error_type=type(exc).__name__,
+            error=repr(exc),
+        )

api/tests/factories.py

@@ -3,7 +3,7 @@
 - `make_osv_entry` — минимальная цепочка FK поверх сессии (`flush`, без `commit`): откатывается фикстурой `db_session`.
 - `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые
   работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется
-  `cleanup()` в порядке FK.
+  `cleanup()` в порядке FK, включая строки `s3_object` файлов результата, зарегистрированные inbox'ом.
 """
 
 import datetime
@@ -11,7 +11,7 @@ import decimal
 import uuid
 from typing import Any
 
-from sqlalchemy import delete
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.models.directories import LegalEntity
@@ -92,6 +92,7 @@ class CommittedJobs:
         self.job_ids: list[uuid.UUID] = []
         self._file_ids: list[uuid.UUID] = []
         self._s3_ids: list[uuid.UUID] = []
+        self._extra_s3_ids: list[uuid.UUID] = []
         self._entity_ids: list[uuid.UUID] = []
 
     async def make_queued_job(  # noqa: PLR0913 - каждое поле задания задаётся тестом явно
@@ -106,6 +107,8 @@ class CommittedJobs:
         request_msg_id: str | None = None,
         metrics: Any = None,
         attempt: int = 1,
+        result_msg_id: str | None = None,
+        started_at: datetime.datetime | None = None,
     ) -> uuid.UUID:
         """Задание (по умолчанию `queued`, tabular, XLSX) со своей цепочкой FK; возвращает `recognition_job.id`."""
         now = datetime.datetime.now(datetime.UTC)
@@ -142,6 +145,8 @@ class CommittedJobs:
                 published_at=published_at,
                 request_msg_id=request_msg_id,
                 metrics=metrics,
+                result_msg_id=result_msg_id,
+                started_at=started_at,
             )
             session.add(job)
             await session.flush()
@@ -166,18 +171,31 @@ class CommittedJobs:
             assert job is not None
             return job
 
+    def track_s3_object(self, s3_object_id: uuid.UUID) -> None:
+        """Запомнить строку `s3_object`, созданную тестом вне фабрики (удаляется в `cleanup()`)."""
+        self._extra_s3_ids.append(s3_object_id)
+
     async def cleanup(self) -> None:
-        """Удаляет всё созданное в порядке FK."""
+        """Удаляет всё созданное в порядке FK (файлы результата, на которые ссылаются задания, — после заданий)."""
         async with self._session_factory() as session, session.begin():
+            result_s3_ids: list[uuid.UUID] = []
             if self.job_ids:
+                rows = await session.execute(
+                    select(RecognitionJob.result_s3_object_id).where(
+                        RecognitionJob.id.in_(self.job_ids), RecognitionJob.result_s3_object_id.is_not(None)
+                    )
+                )
+                result_s3_ids = [row[0] for row in rows]
                 await session.execute(delete(RecognitionJob).where(RecognitionJob.id.in_(self.job_ids)))
             if self._file_ids:
                 await session.execute(delete(StoredFile).where(StoredFile.id.in_(self._file_ids)))
-            if self._s3_ids:
-                await session.execute(delete(S3Object).where(S3Object.id.in_(self._s3_ids)))
+            s3_ids = [*self._s3_ids, *self._extra_s3_ids, *result_s3_ids]
+            if s3_ids:
+                await session.execute(delete(S3Object).where(S3Object.id.in_(s3_ids)))
             if self._entity_ids:
                 await session.execute(delete(LegalEntity).where(LegalEntity.id.in_(self._entity_ids)))
         self.job_ids.clear()
         self._file_ids.clear()
         self._s3_ids.clear()
+        self._extra_s3_ids.clear()
         self._entity_ids.clear()

api/tests/test_app.py

@@ -199,6 +199,31 @@ def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest.
     assert not nats.is_connected  # создание не подключает: подключение — в start()
 
 
+async def test_result_subscription_uses_inbox_handler_with_dependencies():
+    """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения и `NoopMapper`."""
+    subscribed: list[Any] = []
+
+    class _FakeConnection:
+        async def subscribe_results(self, handler: Any) -> None:
+            subscribed.append(handler)
+
+    await app_module._subscribe_results(_FakeConnection())
+
+    (handler,) = subscribed
+    assert handler.func is recognition_exchange.handle_message
+    assert handler.args == ()
+    assert set(handler.keywords) == {"session_factory", "mapper"}
+    assert handler.keywords["session_factory"] is app_module.async_session_maker
+    assert isinstance(handler.keywords["mapper"], recognition_exchange.NoopMapper)
+
+
+def test_create_nats_subscribes_results_on_connect(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
+    nats = app_module._create_nats()
+    assert nats is not None
+    assert nats._on_connected is app_module._subscribe_results
+
+
 @pytest.mark.parametrize("secure", [True, False])
 async def test_connect_s3_passes_secure_flag(monkeypatch: pytest.MonkeyPatch, secure: bool):
     created: dict[str, Any] = {}

api/tests/test_recognition_inbox.py

@@ -0,0 +1,814 @@
+"""Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в
+`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22).
+
+Быстрые тесты (без маркера) — разбор заголовка `Nats-Msg-Id`, слияние метрик, отказ невалидным сообщениям без
+обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён.
+Интеграционные — `handle_message` с фейковым Msg (`SimpleNamespace(data, headers, subject)`) на реальном PostgreSQL
+(БД сессии pytest; данные фиксируются `CommittedJobs` и удаляются в teardown) и полный цикл через реальный NATS
+(уникальный result-subject: общий `recognition.result` не трогаем).
+"""
+
+import asyncio
+import importlib.util
+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 func, select
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+
+from src.contract import NATS_MSG_ID_HEADER, RecognitionResultMessage, ack_msg_id, result_msg_id
+from src.core.config import get_settings
+from src.core.nats import NatsConnection
+from src.models.enums import RecognitionStatus, S3BucketRole
+from src.models.files import RecognitionJob, S3Object
+from src.services.recognition_exchange import (
+    FINAL_STATUSES,
+    NoopMapper,
+    expire_deadlines,
+    handle_message,
+    header_msg_id,
+    merge_metrics,
+    publish_pending,
+)
+from tests.factories import CommittedJobs
+
+API_DIR = Path(__file__).resolve().parents[1]
+MESSAGE_SCHEMA = json.loads((API_DIR / "contracts" / "recognition-result-message.schema.json").read_text("utf-8"))
+MESSAGE_VALIDATOR = Draft202012Validator(MESSAGE_SCHEMA, format_checker=FormatChecker())
+
+SERVICE = {"name": "tabular-parser", "version": "0.1.0"}
+BUCKET = "arm-ka"
+READY_TIMEOUT = 5.0
+RACE_JOBS = 10
+LOCK_HOLD_SECONDS = 0.5
+
+
+# --- инфраструктура ---------------------------------------------------------------------------------------------
+
+
+@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, **extra: Any) -> dict[str, Any]:
+    """Конверт `recognition-result-message` v1.0; для done/needs_check по умолчанию — `result_s3` и версия схемы."""
+    body: dict[str, Any] = {
+        "schema_version": "1.0",
+        "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"] = "1.0"
+    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_VALIDATOR.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="recognition.result")
+
+
+class RecordingMapper:
+    """Подставной mapper: запоминает вызовы и видимое состояние задания; `fail=True` — исключение; `delay` — пауза
+    внутри транзакции (строка задания остаётся заблокированной)."""
+
+    def __init__(self, *, fail: bool = False, delay: float = 0) -> None:
+        self.fail = fail
+        self.delay = delay
+        self.calls: list[tuple[uuid.UUID, RecognitionStatus, RecognitionResultMessage]] = []
+
+    async def map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None:
+        self.calls.append((job.id, job.status, message))
+        if self.delay:
+            await asyncio.sleep(self.delay)
+        if self.fail:
+            raise RuntimeError("подставная ошибка маппинга")
+
+
+@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_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 / JSON null (так ORM пишет None)
+        ([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="2.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="recognition.result")
+    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_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
+
+
+async def test_noop_mapper_does_nothing():
+    message = RecognitionResultMessage.model_validate(envelope(_JOB, _FILE, "done"))
+    assert await NoopMapper().map_result(None, None, message) is None  # type: ignore[arg-type]
+
+
+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_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")
+    assert job["metrics"] == {"republish_count": 1, **metrics}
+    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
+async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
+    """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object не меняются; mapper не вызывается."""
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    msg = fake_msg(envelope(job_id, file_id, "done", metrics={"rows": 3}), result_msg_id(job_id, 1))
+    mapper = RecordingMapper()
+    await _handle(session_factory, msg, mapper=mapper)
+    job_before = await _row(session_factory, job_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, mapper=mapper, 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 len(mapper.calls) == 1
+    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)
+    mapper = RecordingMapper()
+
+    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper)
+
+    assert await _row(session_factory, job_id) == before
+    assert await _s3_total(session_factory) == total_before
+    assert mapper.calls == []
+    (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
+
+
+# --- mapper ------------------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_mapper_sees_job_and_message(session_factory, jobs: CommittedJobs):
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    mapper = RecordingMapper()
+
+    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper)
+
+    ((seen_job_id, seen_status, message),) = mapper.calls
+    assert seen_job_id == job_id
+    assert seen_status is RecognitionStatus.done  # mapper видит уже обновлённое задание (может его уточнить)
+    assert message.job_id == job_id
+    assert message.result_s3 is not None
+    assert message.result_s3.key == _result_key(file_id, job_id)
+
+
+@pytest.mark.integration
+async def test_mapper_error_rolls_back_whole_transaction(session_factory, jobs: CommittedJobs, logs: list[Any]):
+    """Исключение mapper'а — откат всего: задание не изменено, s3_object не создан (дойдёт до timeout, D-6)."""
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    before = await _row(session_factory, job_id)
+    total_before = await _s3_total(session_factory)
+
+    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
+    await _handle(session_factory, msg, mapper=RecordingMapper(fail=True))
+
+    assert await _row(session_factory, job_id) == before
+    assert await _s3_total(session_factory) == total_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]):
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
+    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)  # первый держит строку, второй ждёт его commit
+
+    await asyncio.gather(*[_handle(session_factory, msg, mapper=mapper) 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 len(mapper.calls) == 1
+    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), либо
+    timeout (без результата), но не смесь. Порядок решает блокировка строки: кто первый, тот и финальный."""
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    job_ids = [
+        await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past) for _ in range(RACE_JOBS)
+    ]
+    messages = [
+        fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1)) for job_id in job_ids
+    ]
+
+    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))
+        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"]]
+        else:
+            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):
+    """Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой
+    версии строки (`done`) и задание не трогает."""
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
+    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)
+    msg = fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1))
+    handler = asyncio.create_task(_handle(session_factory, msg, mapper=mapper))
+
+    async def locked() -> None:  # обработчик внутри транзакции, строка заблокирована
+        while not mapper.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
+
+
+@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("recognition.tabular.request", cb=collect)
+        await self._nc.flush()
+
+    async def send(self, body: dict[str, Any], msg_id: str) -> None:
+        MESSAGE_VALIDATOR.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 (результат) → повтор без изменений."""
+    settings = get_settings().OPENBAO.NATS
+    result_subject = f"test.result.{uuid.uuid4().hex}"
+    handler = partial(handle_message, session_factory=session_factory, mapper=NoopMapper())
+
+    async def subscribe(conn: NatsConnection) -> None:
+        await conn.subscribe_results(handler)
+
+    core = NatsConnection(settings, on_connected=subscribe, result_subject=result_subject)
+    job_id = await jobs.make_queued_job()
+    file_id = await _file_id(jobs, job_id)
+    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 (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
+
+        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()