feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)

- services/file_status.py: set_file_status_for_job(session, job, status, *, now, only_from=None) и пакетный
  set_file_status_for_jobs для вахтёра; переход только при current_recognition_job_id = job.id, PROCESSED и
  SUPERSEDED не перезаписываются (D-5, R-5).
- inbox: ack -> файл UPLOADED->PROCESSING; error -> ERROR; done/needs_check -> UPLOADED->PROCESSING (R-4: ack мог
  потеряться), из PROCESSING файл ждёт цикла маппинга; inbox не читает S3 и не создаёт документы.
- вахтёр: UPDATE recognition_job ... RETURNING id и UPDATE stored_file -> ERROR в той же транзакции;
  публикатор при dlq -> файл ERROR.
- удалены ResultMapper, NoopMapper, _map_result, _MapperError и параметр mapper у handle_message; app.py без mapper.
- тесты: статусы файла (ack/error/done/needs_check/без ack/timeout/dlq), чужое задание файл не трогает (Review
  Focus 1), откат статуса файла вместе с транзакцией сообщения; CommittedJobs.make_file_job/add_job/file.
- tasks.md 2.2, шаги Task 4 в плане.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

api/src/app.py

@@ -10,7 +10,7 @@ middleware не подключается), healthcheck (uptime, version, nats: o
 состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и
 вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения
 `recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий
-приложения и `NoopMapper` (маппинг результата в доменные таблицы — Срез 2).
+приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1).
 Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного
 логируется и не отменяет остальные.
 """
@@ -42,7 +42,7 @@ from src.core.config import Settings, get_settings
 from src.core.exception_handlers import register_exception_handlers
 from src.core.nats import MessageHandler, NatsConnection
 from src.db.db import async_session_maker, engine, session_transaction
-from src.services.recognition_exchange import ExchangeTasks, NoopMapper, handle_message
+from src.services.recognition_exchange import ExchangeTasks, handle_message
 from src.services.session_audit import SessionAudit
 
 settings = get_settings()
@@ -102,8 +102,9 @@ async def _connect_s3() -> S3Storage | None:
 
 
 def _result_handler() -> MessageHandler:
-    """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения и `NoopMapper` (Срез 0)."""
+    """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл,
-    return partial(handle_message, session_factory=async_session_maker, mapper=NoopMapper())
+    Срез 2c design D-1)."""
+    return partial(handle_message, session_factory=async_session_maker)
 
 
 async def _subscribe_results(nats: NatsConnection) -> None:

api/src/services/file_status.py

@@ -0,0 +1,85 @@
+"""Статус `stored_file` по заданию распознавания (spec primary-files «Жизненный цикл статуса файла»; design D-5;
+рулинги R-4, R-5; docs/context/05-domain-rules.md §9.2).
+
+Единственное правило перехода: статус файла меняет только его ТЕКУЩЕЕ задание
+(`stored_file.current_recognition_job_id = job.id`). Результат, ack, таймаут или dlq старого задания (повторная загрузка
+после `SUPERSEDED`, будущий replay) файл не трогает. `PROCESSED` и `SUPERSEDED` — итог маппинга и автозамены — обменом
+не перезаписываются.
+
+Функции выполняют `UPDATE` в транзакции вызывающего (inbox, вахтёр, публикатор) — после изменения строки задания:
+порядок блокировок «задание → файл» одинаков у всех участников, взаимоблокировок нет. `audit_event` не пишется:
+обмен не аудируется (ASSUMPTION(OQ-55)).
+"""
+
+import uuid
+from collections.abc import Collection, Sequence
+from datetime import datetime
+from typing import Any, Protocol
+
+from sqlalchemy import CursorResult, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.enums import FileStatus
+from src.models.files import StoredFile
+
+# Итоги маппинга и автозамены: события обмена их не перезаписывают (design D-5)
+PROTECTED_FILE_STATUSES: frozenset[FileStatus] = frozenset({FileStatus.PROCESSED, FileStatus.SUPERSEDED})
+
+
+class JobRef(Protocol):
+    """Что нужно от задания: `recognition_job.id` и `file_id` (ORM `RecognitionJob` или строка выборки)."""
+
+    @property
+    def id(self) -> uuid.UUID: ...
+
+    @property
+    def file_id(self) -> uuid.UUID: ...
+
+
+def _rowcount(result: Any) -> int:
+    return result.rowcount if isinstance(result, CursorResult) else 0
+
+
+async def set_file_status_for_job(
+    session: AsyncSession,
+    job: JobRef,
+    status: FileStatus,
+    *,
+    now: datetime,
+    only_from: Collection[FileStatus] | None = None,
+) -> bool:
+    """Перевести файл задания `job` в `status` (`updated_at = now`), если задание — текущее для файла и файл не в
+    `PROCESSED`/`SUPERSEDED`; `only_from` — допустимые исходные статусы (R-5: ack и результат без ack — только из
+    `UPLOADED`, запоздавший ack не воскрешает `ERROR`). Возвращает `True`, если строка файла изменена."""
+    stmt = (
+        update(StoredFile)
+        .where(
+            StoredFile.id == job.file_id,
+            StoredFile.current_recognition_job_id == job.id,
+            StoredFile.status.not_in(PROTECTED_FILE_STATUSES),
+        )
+        .values(status=status, updated_at=now)
+        .execution_options(synchronize_session=False)
+    )
+    if only_from is not None:
+        stmt = stmt.where(StoredFile.status.in_(list(only_from)))
+    return _rowcount(await session.execute(stmt)) == 1
+
+
+async def set_file_status_for_jobs(
+    session: AsyncSession, job_ids: Sequence[uuid.UUID], status: FileStatus, *, now: datetime
+) -> int:
+    """Пакетный вариант для вахтёра: файлы, у которых текущее задание из `job_ids` и статус не `PROCESSED`/
+    `SUPERSEDED`, → `status`. Возвращает число изменённых файлов."""
+    if not job_ids:
+        return 0
+    stmt = (
+        update(StoredFile)
+        .where(
+            StoredFile.current_recognition_job_id.in_(list(job_ids)),
+            StoredFile.status.not_in(PROTECTED_FILE_STATUSES),
+        )
+        .values(status=status, updated_at=now)
+        .execution_options(synchronize_session=False)
+    )
+    return _rowcount(await session.execute(stmt))

api/src/services/recognition_exchange.py

@@ -4,9 +4,13 @@ docs/context/03-architecture.md, «Контракт обмена через NATS
 Outbox (задача 23):
 - `publish_pending()` — один проход публикатора: `queued` → первая публикация → `published`; `published` без ack дольше
   `ACK_TIMEOUT_SECONDS` → повторная публикация с тем же `Nats-Msg-Id` и тем же `attempt` (счётчик —
-  `metrics.republish_count`); после `OUTBOX_MAX_REPUBLISH` повторов → `dlq`, `error_code = NO_ACK`. `attempt` не растёт:
+  `metrics.republish_count`); после `OUTBOX_MAX_REPUBLISH` повторов → `dlq`, `error_code = NO_ACK`, файл текущего
-  обработка не начиналась. Replay (`attempt + 1`) — вне Среза 0.
+  задания → `ERROR`. `attempt` не растёт: обработка не начиналась. Replay (`attempt + 1`) — вне Среза 2c.
-- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`.
+
+Статусы `stored_file` (Срез 2c; spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5) —
+только через `services/file_status.py` и только для текущего задания файла (`current_recognition_job_id = job.id`).
+- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`;
+  файлы этих заданий (по `current_recognition_job_id`) → `ERROR` в той же транзакции.
 - `run_publisher_loop()` / `run_watchdog_loop()` — фоновые циклы; `ExchangeTasks` запускает их в lifespan при
   `NATS_ENABLED=true` и останавливает до закрытия NATS.
 
@@ -20,22 +24,25 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро
 параллельный проход (второй инстанс ядра) пропускает заблокированную строку и не публикует её повторно, а после commit
 строка уже не подходит под условие выборки.
 
-Обмен не пишет `audit_event` (рулинг R-7, design #20).
+Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)).
 
 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 (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING`
-  тихо игнорируется; ack для другого статуса — warning, задание не меняется.
+  (R-5: только из `UPLOADED`); повторный 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
+  `s3_object` (`bucket_role = results`), обновление задания, статус файла и commit. `error` → файл `ERROR`;
-  заменит маппингом результата в `document` / `osv_entry` / `financial_statement_*`) и commit.
+  `done`/`needs_check` → файл `UPLOADED → PROCESSING` (R-4: ack мог потеряться — core NATS без JetStream), из
+  `PROCESSING` файл не меняется и ждёт цикла маппинга (design D-1). Inbox НЕ читает файл результата и не создаёт
+  документы: `ResultMapper` удалён (OQ-67), маппинг — отдельный цикл `services/result_mapping.py`.
 - Гонка с вахтёром (`expire_deadlines` — один `UPDATE` по статусу) решается блокировкой строки: кто первым взял
   строку, тот и задаёт финальный статус; второй видит финальный статус и ничего не меняет.
-- Невалидное сообщение, неизвестный `job_id`, ошибка БД или mapper'а — лог без повторов; исключения наружу не выходят.
+- Невалидное сообщение, неизвестный `job_id`, ошибка БД — лог без повторов; исключения наружу не выходят.
   Задание без применённого результата дойдёт до `timeout` (осознанный компромисс D-6).
 """
 
@@ -49,7 +56,7 @@ from typing import Any, Protocol
 
 from loguru import logger
 from pydantic import ValidationError
-from sqlalchemy import CursorResult, and_, case, func, literal, or_, select, update
+from sqlalchemy import 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
@@ -69,8 +76,9 @@ from src.contract import (
 )
 from src.core.config import Settings
 from src.core.nats import NatsPublishError
-from src.models.enums import RecognitionPipeline, RecognitionStatus, S3BucketRole
+from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole
 from src.models.files import RecognitionJob, S3Object, StoredFile
+from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs
 
 WATCHDOG_PERIOD_SECONDS = 30  # период вахтёра сроков (spec «Таймаут ожидания результата»)
 PUBLISH_BATCH_SIZE = 50  # заданий за один проход публикатора (design D-6, LIMIT 50)
@@ -220,6 +228,7 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
                 finished_at=now,
             )
         )
+        await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
         report.dlq += 1
         logger.warning(
             "Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, "
@@ -348,8 +357,9 @@ async def publish_pending(  # noqa: PLR0913 - параметры прохода
 
 async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, now: Clock | None = None) -> int:
     """Вахтёр сроков (spec «Таймаут ожидания результата»): `processing`/`published` с `reply_deadline < now` →
-    `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE` с условием по статусу: результат,
+    `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE … RETURNING id` с условием по
-    успевший перевести задание в финальный статус, не перетирается. Возвращает число заданий."""
+    статусу: результат, успевший перевести задание в финальный статус, не перетирается. В той же транзакции файлы этих
+    заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`. Возвращает число заданий."""
     moment = (now or _utcnow)()
     stmt = (
         update(RecognitionJob)
@@ -363,13 +373,19 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *,
             error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)",
             finished_at=moment,
         )
+        .returning(RecognitionJob.id)
         .execution_options(synchronize_session=False)
     )
     async with session_factory() as session, session.begin():
-        result = await session.execute(stmt)
+        job_ids = list((await session.execute(stmt)).scalars())
-    count = result.rowcount if isinstance(result, CursorResult) else 0
+        files = await set_file_status_for_jobs(session, job_ids, FileStatus.ERROR, now=moment)
+    count = len(job_ids)
     if count:
-        logger.warning("Задания распознавания переведены в timeout по сроку ответа. count={count}", count=count)
+        logger.warning(
+            "Задания распознавания переведены в timeout по сроку ответа. count={count}, files_error={files}",
+            count=count,
+            files=files,
+        )
     return count
 
 
@@ -495,31 +511,6 @@ class InboundMessage(Protocol):
     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:
@@ -588,19 +579,26 @@ async def _lock_job(session: AsyncSession, job_id: uuid.UUID) -> RecognitionJob
     return (await session.execute(stmt)).scalar_one_or_none()
 
 
-def _apply_ack(job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime) -> None:
+async def _apply_ack(
-    """Ack (`status = processing`): `published` → `processing`; повтор для `processing` — тихо; иначе — warning."""
+    session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime
+) -> None:
+    """Ack (`status = processing`): `published` → `processing` и файл `UPLOADED → PROCESSING` (R-5); повтор для
+    `processing` — тихо; иначе — warning (запоздавший ack после `timeout`/`dlq` файл не трогает)."""
     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
+        file_changed = await set_file_status_for_job(
+            session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED}
+        )
         logger.info(
             "Сервис подтвердил получение запроса (ack): задание в processing. job_id={job_id}, service={service}, "
-            "msg_id={msg_id}",
+            "msg_id={msg_id}, file_processing={file_changed}",
             job_id=job.id,
             service=message.service.name.value,
             msg_id=msg_id,
+            file_changed=file_changed,
         )
     elif job.status is RecognitionStatus.processing:
         logger.debug("Повторный ack для задания в processing, игнорируется. job_id={job_id}", job_id=job.id)
@@ -653,7 +651,7 @@ async def _apply_result(
     session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime
 ) -> bool:
     """Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе —
-    регистрация файла результата и обновление задания (`True`, дальше — mapper и commit)."""
+    регистрация файла результата, обновление задания и статуса файла (`True`, дальше — commit)."""
     if job.result_msg_id == msg_id:
         logger.debug(
             "Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, "
@@ -689,26 +687,26 @@ async def _apply_result(
     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
+    file_changed = await _apply_result_file_status(session, job, now)
-
-
-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}",
+        "error_code={error_code}, file_changed={file_changed}",
         job_id=job.id,
         status=job.status.value,
         service=message.service.name.value,
-        msg_id=job.result_msg_id,
+        msg_id=msg_id,
         error_code=job.error_code,
+        file_changed=file_changed,
     )
+    return True
+
+
+async def _apply_result_file_status(session: AsyncSession, job: RecognitionJob, now: datetime) -> bool:
+    """Статус файла по принятому результату (design D-5): `error` → `ERROR`; `done`/`needs_check` → `PROCESSING` только
+    из `UPLOADED` (R-4: ack мог потеряться), иначе файл ждёт цикла маппинга как есть. Только текущее задание файла."""
+    if job.status is RecognitionStatus.error:
+        return await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
+    return await set_file_status_for_job(session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED})
 
 
 def _metrics_json(metrics: Metrics) -> dict[str, Any]:
@@ -716,7 +714,7 @@ def _metrics_json(metrics: Metrics) -> dict[str, Any]:
 
 
 async def _apply_message(
-    session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, mapper: ResultMapper, clock: Clock
+    session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, clock: Clock
 ) -> None:
     job = await _lock_job(session, message.job_id)
     if job is None:
@@ -748,36 +746,30 @@ async def _apply_message(
         )
     now = clock()
     if is_ack:
-        _apply_ack(job, message, msg_id, now)
+        await _apply_ack(session, job, message, msg_id, now)
-    elif await _apply_result(session, job, message, msg_id, now):
+    else:
-        await _map_result(session, job, message, mapper)
+        await _apply_result(session, job, message, msg_id, now)
 
 
 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'а — откат транзакции,
+    Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл),
-    `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7).
+    `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7,
+    ASSUMPTION(OQ-55)). Файл результата не читается: маппинг — отдельный цикл (design D-1).
     """
     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)
+            await _apply_message(session, message, header_msg_id(msg.headers), 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, "

api/tests/factories.py

@@ -8,7 +8,8 @@
   `FakeStorage` для тестов загрузки и маппинга (Срез 2c; `flush`, без `commit`). `make_entries_from_result` создаёт
   Task 9 (R-2).
 - `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые
-  работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется
+  работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. `make_file_job` / `add_job` — файл
+  с текущим заданием и второе задание того же файла (статусы `stored_file` в обмене, Срез 2c). Всё созданное удаляется
   `cleanup()` в порядке FK, включая строки `s3_object` файлов результата, зарегистрированные inbox'ом.
 """
 
@@ -400,6 +401,71 @@ class CommittedJobs:
             self.job_ids.append(job.id)
             return job.id
 
+    async def make_file_job(
+        self,
+        *,
+        file_status: FileStatus = FileStatus.UPLOADED,
+        job_status: RecognitionStatus = RecognitionStatus.queued,
+        **job_fields: Any,
+    ) -> tuple[uuid.UUID, uuid.UUID]:
+        """Файл с текущим заданием (`make_stored_file`: `current_recognition_job_id` = задание), зафиксированный в БД.
+
+        `job_fields` — поля `recognition_job` поверх фабрики (`reply_deadline`, `published_at`, `metrics`, …).
+        Возвращает `(stored_file.id, recognition_job.id)`.
+        """
+        async with self._session_factory() as session, session.begin():
+            entity = await make_legal_entity(session, name="ООО Тест статусов файла")
+            file, job = await make_stored_file(session, entity=entity, status=file_status, job_status=job_status)
+            for name, value in job_fields.items():
+                setattr(job, name, value)
+            await session.flush()
+            self._entity_ids.append(entity.id)
+            self._s3_ids.append(file.s3_object_id)
+            self._file_ids.append(file.id)
+            self.job_ids.append(job.id)
+            return file.id, job.id
+
+    async def add_job(
+        self,
+        file_id: uuid.UUID,
+        *,
+        status: RecognitionStatus = RecognitionStatus.queued,
+        current: bool = True,
+        **job_fields: Any,
+    ) -> uuid.UUID:
+        """Ещё одно задание файла `file_id` (как replay: `attempt` = максимум по файлу + 1); `current=True` — оно
+        становится текущим заданием файла."""
+        async with self._session_factory() as session, session.begin():
+            file = await session.get(StoredFile, file_id)
+            assert file is not None
+            last = (
+                await session.execute(select(func.max(RecognitionJob.attempt)).where(RecognitionJob.file_id == file_id))
+            ).scalar_one()
+            job_fields.setdefault("attempt", (last or 0) + 1)
+            tabular = file.file_type in TABULAR_FILE_TYPES
+            pipeline = RecognitionPipeline.tabular if tabular else RecognitionPipeline.image
+            job = RecognitionJob(
+                file_id=file_id,
+                pipeline=pipeline,
+                request_subject=f"recognition.{pipeline.value}.request",
+                schema_version="1.0",
+                status=status,
+                **job_fields,
+            )
+            session.add(job)
+            await session.flush()
+            if current:
+                file.current_recognition_job_id = job.id
+            self.job_ids.append(job.id)
+            return job.id
+
+    async def file(self, file_id: uuid.UUID) -> StoredFile:
+        """Свежее состояние файла из БД (новая сессия)."""
+        async with self._session_factory() as session:
+            file = await session.get(StoredFile, file_id)
+            assert file is not None
+            return file
+
     async def s3_key(self, job_id: uuid.UUID) -> str:
         """Ключ S3 исходного файла задания."""
         async with self._session_factory() as session:

api/tests/test_app.py

@@ -226,7 +226,8 @@ def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest.
 
 
 async def test_result_subscription_uses_inbox_handler_with_dependencies():
-    """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения и `NoopMapper`."""
+    """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения (без `mapper`: маппинг
+    результата — отдельный цикл, Срез 2c D-1)."""
     subscribed: list[Any] = []
 
     class _FakeConnection:
@@ -238,9 +239,8 @@ async def test_result_subscription_uses_inbox_handler_with_dependencies():
     (handler,) = subscribed
     assert handler.func is recognition_exchange.handle_message
     assert handler.args == ()
-    assert set(handler.keywords) == {"session_factory", "mapper"}
+    assert set(handler.keywords) == {"session_factory"}
     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):

api/tests/test_exchange_lifespan_nats.py

@@ -32,7 +32,7 @@ from src.core.config import get_settings
 from src.core.nats import RESULT_QUEUE_GROUP, NatsConnection
 from src.models.enums import RecognitionStatus, S3BucketRole
 from src.models.files import RecognitionJob, S3Object
-from src.services.recognition_exchange import NoopMapper, handle_message
+from src.services.recognition_exchange import handle_message
 from tests.factories import CommittedJobs
 from tests.test_recognition_inbox import MESSAGE_VALIDATOR, envelope
 
@@ -277,10 +277,7 @@ async def test_two_core_instances_handle_result_once(
     """Два `NatsConnection` в queue group `arm-core` на одном subject: результат обрабатывает ровно один инстанс,
     задание меняется один раз; повторная доставка того же сообщения ничего не меняет."""
     settings = get_settings().OPENBAO.NATS
-    handlers = [
+    handlers = [_CountingHandler(partial(handle_message, session_factory=session_factory)) for _ in range(2)]
-        _CountingHandler(partial(handle_message, session_factory=session_factory, mapper=NoopMapper()))
-        for _ in range(2)
-    ]
 
     def subscriber(handler: _CountingHandler) -> Callable[[NatsConnection], Awaitable[None]]:
         async def subscribe(conn: NatsConnection) -> None:

api/tests/test_recognition_exchange.py

@@ -1,5 +1,7 @@
 """Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из
-`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6).
+`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). Срез 2c:
+`timeout` вахтёра и `dlq` публикатора переводят файл текущего задания в `ERROR` (spec primary-files «Жизненный цикл
+статуса файла», design D-5).
 
 Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и
 вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown.
@@ -28,7 +30,7 @@ from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest
 from src.contract.request import FileType as ContractFileType
 from src.core.config import get_settings
 from src.core.nats import NatsConnection, NatsPublishError
-from src.models.enums import FileType, RecognitionPipeline, RecognitionStatus
+from src.models.enums import FileStatus, FileType, RecognitionPipeline, RecognitionStatus
 from src.models.files import RecognitionJob
 from src.services import recognition_exchange as rx
 from src.services.recognition_exchange import (
@@ -591,6 +593,93 @@ async def test_expire_deadlines(
         assert job.finished_at is None
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED])
+@pytest.mark.parametrize("job_status", [RecognitionStatus.processing, RecognitionStatus.published])
+async def test_expire_deadlines_moves_current_file_to_error(
+    session_factory: async_sessionmaker[AsyncSession],
+    jobs: CommittedJobs,
+    file_status: FileStatus,
+    job_status: RecognitionStatus,
+):
+    """Сценарий «Сервис не ответил» (primary-files): задание — `timeout`, файл — `ERROR` в той же транзакции."""
+    clock = Clock()
+    file_id, job_id = await jobs.make_file_job(
+        file_status=file_status, job_status=job_status, reply_deadline=clock() - timedelta(minutes=1)
+    )
+
+    assert await expire_deadlines(session_factory, now=clock) == 1
+
+    job, file = await jobs.get(job_id), await jobs.file(file_id)
+    assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")
+    assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
+async def test_expire_deadlines_keeps_processed_and_superseded_file(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, file_status: FileStatus
+):
+    clock = Clock()
+    file_id, job_id = await jobs.make_file_job(
+        file_status=file_status, job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1)
+    )
+
+    assert await expire_deadlines(session_factory, now=clock) == 1
+
+    assert (await jobs.get(job_id)).status is RecognitionStatus.timeout
+    file = await jobs.file(file_id)
+    assert (file.status, file.updated_at) == (file_status, None)
+
+
+@pytest.mark.integration
+async def test_expire_deadlines_of_non_current_job_keeps_file(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
+):
+    """Review Focus 1: таймаут старого задания файла, у которого уже новое текущее задание, файл не трогает."""
+    clock = Clock()
+    file_id, old_job = await jobs.make_file_job(
+        file_status=FileStatus.UPLOADED,
+        job_status=RecognitionStatus.processing,
+        reply_deadline=clock() - timedelta(minutes=1),
+    )
+    new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)
+
+    assert await expire_deadlines(session_factory, now=clock) == 1
+
+    assert (await jobs.get(old_job)).status is RecognitionStatus.timeout
+    file = await jobs.file(file_id)
+    assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
+async def test_dlq_moves_current_file_to_error(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool
+):
+    """Публикатор переводит задание в `dlq` — файл текущего задания `ERROR`; файл чужого задания не меняется."""
+    clock = Clock()
+    file_id, job_id = await jobs.make_file_job(
+        job_status=RecognitionStatus.published,
+        published_at=clock() - ACK * 2,
+        request_msg_id="dlq-test:1",
+        reply_deadline=clock() + timedelta(hours=1),
+        metrics={"republish_count": MAX_REPUBLISH},
+    )
+    # новое задание — `processing`, чтобы публикатор его не трогал (в проходе только задание старого)
+    new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
+
+    report = await _publish(session_factory, FakeNats(), clock)
+
+    assert report.dlq == 1
+    assert (await jobs.get(job_id)).status is RecognitionStatus.dlq
+    file = await jobs.file(file_id)
+    if current:
+        assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
+    else:
+        assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
+
+
 # --- фоновые циклы -----------------------------------------------------------------------------------------------
 
 

api/tests/test_recognition_inbox.py

@@ -1,5 +1,7 @@
 """Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в
-`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22).
+`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` удалён.
@@ -10,6 +12,7 @@
 
 import asyncio
 import importlib.util
+import inspect
 import json
 import time
 import uuid
@@ -29,14 +32,14 @@ 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, RecognitionResultMessage, ack_msg_id, result_msg_id
+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 RecognitionStatus, S3BucketRole
+from src.models.enums import FileStatus, RecognitionStatus, S3BucketRole
-from src.models.files import RecognitionJob, S3Object
+from src.models.files import RecognitionJob, S3Object, StoredFile
+from src.services import recognition_exchange as rx
 from src.services.recognition_exchange import (
     FINAL_STATUSES,
-    NoopMapper,
     expire_deadlines,
     handle_message,
     header_msg_id,
@@ -54,6 +57,7 @@ BUCKET = "arm-ka"
 READY_TIMEOUT = 5.0
 RACE_JOBS = 10
 LOCK_HOLD_SECONDS = 0.5
+_ERROR = {"code": "UNSUPPORTED_FORMAT", "message": "формат файла не поддерживается"}
 
 
 # --- инфраструктура ---------------------------------------------------------------------------------------------
@@ -103,21 +107,20 @@ def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER,
     return SimpleNamespace(data=data, headers=headers, subject="recognition.result")
 
 
-class RecordingMapper:
+class SlowFileStatus:
-    """Подставной mapper: запоминает вызовы и видимое состояние задания; `fail=True` — исключение; `delay` — пауза
+    """Подмена `set_file_status_for_job` в inbox: запоминает вызовы (id задания, статус) и держит паузу `delay`
-    внутри транзакции (строка задания остаётся заблокированной)."""
+    ВНУТРИ транзакции сообщения (строка задания остаётся заблокированной), затем вызывает настоящую функцию."""
 
-    def __init__(self, *, fail: bool = False, delay: float = 0) -> None:
+    def __init__(self, monkeypatch: pytest.MonkeyPatch, *, delay: float) -> None:
-        self.fail = fail
         self.delay = delay
-        self.calls: list[tuple[uuid.UUID, RecognitionStatus, RecognitionResultMessage]] = []
+        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 map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None:
+    async def __call__(self, session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool:
-        self.calls.append((job.id, job.status, message))
+        self.calls.append((job.id, status))
-        if self.delay:
         await asyncio.sleep(self.delay)
-        if self.fail:
+        return await self._original(session, job, status, **kwargs)
-            raise RuntimeError("подставная ошибка маппинга")
 
 
 @pytest.fixture
@@ -152,6 +155,13 @@ async def _s3_total(session_factory: async_sessionmaker[AsyncSession]) -> int:
         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
 
@@ -264,9 +274,11 @@ async def test_database_error_is_logged_and_not_raised(logs: list[Any]):
     assert record["exception"] is not None
 
 
-async def test_noop_mapper_does_nothing():
+def test_inbox_has_no_result_mapper():
-    message = RecognitionResultMessage.model_validate(envelope(_JOB, _FILE, "done"))
+    """OQ-67 (Срез 2c, D-1): точки расширения `ResultMapper` / `NoopMapper` нет, `handle_message` без `mapper`."""
-    assert await NoopMapper().map_result(None, None, message) is None  # type: ignore[arg-type]
+    assert not hasattr(rx, "ResultMapper")
+    assert not hasattr(rx, "NoopMapper")
+    assert "mapper" not in inspect.signature(handle_message).parameters
 
 
 def test_final_statuses():
@@ -422,22 +434,23 @@ async def test_result_metrics_merge_over_null_metrics(
 
 @pytest.mark.integration
 async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
-    """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object не меняются; mapper не вызывается."""
+    """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object, ни stored_file не меняются."""
-    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id, job_id = await jobs.make_file_job(
-    file_id = await _file_id(jobs, job_id)
+        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))
-    mapper = RecordingMapper()
+    await _handle(session_factory, msg)
-    await _handle(session_factory, msg, mapper=mapper)
     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, mapper=mapper, now=lambda: datetime.now(UTC) + timedelta(hours=1))
+    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 len(mapper.calls) == 1
+    assert await _file_row(session_factory, file_id) == file_before
     assert _logged(logs, "дубликат", "DEBUG")
     assert not _logged(logs, "финальном статусе")
 
@@ -482,13 +495,11 @@ async def test_result_for_final_status_is_logged_and_ignored(
     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)
+    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
-    assert mapper.calls == []
     (record,) = _logged(logs, "финальном статусе", "WARNING")
     assert status.value in record["message"]
 
@@ -587,39 +598,163 @@ async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs
     assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1
 
 
-# --- mapper ------------------------------------------------------------------------------------------------------
+# --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) -------------
 
 
 @pytest.mark.integration
-async def test_mapper_sees_job_and_message(session_factory, jobs: CommittedJobs):
+async def test_ack_moves_uploaded_file_to_processing(session_factory, jobs: CommittedJobs):
-    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    """Сценарий «Сервис подтвердил»: файл текущего задания — `PROCESSING`, `updated_at` — момент ack."""
-    file_id = await _file_id(jobs, job_id)
+    file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.published)
-    mapper = RecordingMapper()
+    moment = datetime(2026, 10, 5, 9, 0, tzinfo=UTC)
 
-    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper)
+    await _handle(
+        session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
+    )
 
-    ((seen_job_id, seen_status, message),) = mapper.calls
+    file = await _file_row(session_factory, file_id)
-    assert seen_job_id == job_id
+    assert file["status"] is FileStatus.PROCESSING
-    assert seen_status is RecognitionStatus.done  # mapper видит уже обновлённое задание (может его уточнить)
+    assert file["updated_at"] == moment
-    assert message.job_id == job_id
+    assert file["current_recognition_job_id"] == job_id
-    assert message.result_s3 is not None
+    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing
-    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]):
+@pytest.mark.parametrize(
-    """Исключение mapper'а — откат всего: задание не изменено, s3_object не создан (дойдёт до timeout, D-6)."""
+    "file_status", [FileStatus.ERROR, FileStatus.PROCESSED, FileStatus.SUPERSEDED, FileStatus.PROCESSING]
-    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+)
-    file_id = await _file_id(jobs, job_id)
+async def test_ack_moves_file_only_from_uploaded(session_factory, jobs: CommittedJobs, file_status: FileStatus):
-    before = await _row(session_factory, job_id)
+    """R-5: ack ставит `PROCESSING` только из `UPLOADED` — запоздавший ack не воскрешает `ERROR` и не трогает
-    total_before = await _s3_total(session_factory)
+    `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)
 
-    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
+    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))
-    await _handle(session_factory, msg, mapper=RecordingMapper(fail=True))
 
-    assert await _row(session_factory, job_id) == before
+    assert await _file_row(session_factory, file_id) == before
-    assert await _s3_total(session_factory) == total_before
+    assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing
-    (record,) = _logged(logs, "Ошибка маппинга", "ERROR")
+
+
+@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
 
 
@@ -627,17 +762,19 @@ async def test_mapper_error_rolls_back_whole_transaction(session_factory, jobs:
 
 
 @pytest.mark.integration
-async def test_concurrent_same_message_applied_once(session_factory, jobs: CommittedJobs, logs: list[Any]):
+async def test_concurrent_same_message_applied_once(
-    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch
-    file_id = await _file_id(jobs, job_id)
+):
+    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))
-    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)  # первый держит строку, второй ждёт его commit
+    slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS)  # первый держит строку, второй ждёт его commit
 
-    await asyncio.gather(*[_handle(session_factory, msg, mapper=mapper) for _ in range(2)])
+    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 len(mapper.calls) == 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, "неизвестного задания")
 
@@ -684,26 +821,32 @@ async def test_ack_waits_for_publisher_lock(session_factory, jobs: CommittedJobs
 
 @pytest.mark.integration
 async def test_result_vs_watchdog_is_consistent(session_factory, jobs: CommittedJobs):
-    """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object), либо
+    """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object, файл
-    timeout (без результата), но не смесь. Порядок решает блокировка строки: кто первый, тот и финальный."""
+    `PROCESSING`), либо timeout (без результата, файл `ERROR`), но не смесь. Порядок решает блокировка строки задания:
+    кто первый, тот и финальный (и задание, и файл меняются в порядке «задание → файл», без взаимоблокировок)."""
     past = datetime.now(UTC) - timedelta(minutes=1)
-    job_ids = [
+    pairs = [
-        await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past) for _ in range(RACE_JOBS)
+        await jobs.make_file_job(
-    ]
+            file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
-    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
+        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
@@ -712,17 +855,21 @@ async def test_result_vs_watchdog_is_consistent(session_factory, jobs: Committed
 
 
 @pytest.mark.integration
-async def test_watchdog_waits_for_result_transaction(session_factory, jobs: CommittedJobs):
+async def test_watchdog_waits_for_result_transaction(
+    session_factory, jobs: CommittedJobs, monkeypatch: pytest.MonkeyPatch
+):
     """Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой
-    версии строки (`done`) и задание не трогает."""
+    версии строки (`done`) и задание не трогает; файл остаётся `PROCESSING` (не `ERROR`)."""
     past = datetime.now(UTC) - timedelta(minutes=1)
-    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
+    file_id, job_id = await jobs.make_file_job(
-    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)
+        file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
-    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))
+    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 mapper.calls:
+        while not slow.calls:
             await asyncio.sleep(0.01)
 
     await asyncio.wait_for(locked(), 5)
@@ -734,6 +881,7 @@ async def test_watchdog_waits_for_result_transaction(session_factory, jobs: Comm
     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
@@ -805,17 +953,17 @@ class _FakeService:
 
 @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 (результат) → повтор без изменений."""
+    """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, mapper=NoopMapper())
+    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)
-    job_id = await jobs.make_queued_job()
+    file_id, job_id = await jobs.make_file_job()
-    file_id = await _file_id(jobs, job_id)
     service = _FakeService(job_id, result_subject)
     await core.start()
     try:
@@ -834,6 +982,7 @@ async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, l
         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})
@@ -843,6 +992,7 @@ async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, l
         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))
 

docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md

@@ -137,9 +137,9 @@ async def set_file_status_for_job(session, job: RecognitionJob, status: FileStat
 ```
 Ошибка задания (`error`/`timeout`/`dlq`) ставит файлу `ERROR`; ack — `PROCESSING` только из `UPLOADED`; `done`/`needs_check` файл не трогают.
 
-- [ ] **Step 1: Тесты** (integration, `CommittedJobs` + `make_stored_file`): ack → `PROCESSING`; `error` → `ERROR`; `done` → файл `PROCESSING`; `timeout` вахтёром → `ERROR`; `dlq` публикатором → `ERROR`; результат задания, которое уже не `current_recognition_job_id` файла → файл не меняется (Review Focus 1); существующие тесты inbox без `mapper` зелёные.
+- [x] **Step 1: Тесты** (integration, `CommittedJobs` + `make_stored_file`): ack → `PROCESSING`; `error` → `ERROR`; `done` → файл `PROCESSING`; `timeout` вахтёром → `ERROR`; `dlq` публикатором → `ERROR`; результат задания, которое уже не `current_recognition_job_id` файла → файл не меняется (Review Focus 1); существующие тесты inbox без `mapper` зелёные.
-- [ ] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q`, `make check`.
+- [x] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q`, `make check`.
-- [ ] **Step 5: Commit** `feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)`.
+- [x] **Step 5: Commit** `feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)`.
 
 ---
 

openspec/changes/slice-2c-osv-ingest/tasks.md

@@ -23,7 +23,7 @@
       `schemas/files.py` (`UploadResult{accepted,rejected}`, `FileOut`, `JobBrief`, `DocumentBrief`), эндпоинт
       `POST /legal-entities/{id}/files`, роутер. Тесты: сценарии спеки `primary-files` (принята, смесь, сигнатура, чужое ЮЛ, дубль);
       `fixtures`-тест: реальный xls принят и `publish_pending` на `FakeNats` публикует запрос с ключом `uploads/{file_id}/{job_id}/…`.
-- [ ] 2.2 **Статусы файла в обмене.** `recognition_exchange.py`: ack → `PROCESSING`; `error` → `ERROR`; вахтёр `timeout` → `ERROR`;
+- [x] 2.2 **Статусы файла в обмене.** `recognition_exchange.py`: ack → `PROCESSING`; `error` → `ERROR`; вахтёр `timeout` → `ERROR`;
       публикатор `dlq` → `ERROR` (все — только при `current_recognition_job_id = job`); `ResultMapper`/`NoopMapper` удалены,
       `_map_result` убран, тесты inbox обновлены (`RecordingMapper` → проверка статуса файла). Тесты: сценарии спеки
       `recognition-job-exchange` (ack → PROCESSING, error → ERROR, timeout → ERROR, done → файл остаётся PROCESSING).