fix(exchange): INVALID_REQUEST публикатора переводит файл в ERROR

T4.1: публикатор, переводя задание в финальный error/INVALID_REQUEST, теперь в той же транзакции (под тем же
FOR UPDATE) ставит файлу текущего задания ERROR (spec primary-files «Жизненный цикл статуса файла»); файл
чужого задания не меняется. Тест test_invalid_request_marks_job_error_and_pass_continues[current|non_current];
CommittedJobs.make_file_job принимает original_name.

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

api/src/services/recognition_exchange.py

@@ -251,6 +251,8 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
                 finished_at=now,
             )
         )
+        # финальная ошибка задания — файл текущего задания ERROR (spec primary-files «Жизненный цикл статуса файла»)
+        await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
         report.invalid += 1
         logger.error(
             "Запрос распознавания не собран по контракту, задание переведено в error. "

api/tests/factories.py

@@ -406,16 +406,25 @@ class CommittedJobs:
         *,
         file_status: FileStatus = FileStatus.UPLOADED,
         job_status: RecognitionStatus = RecognitionStatus.queued,
+        original_name: str | None = "osv.xls",
         **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)`.
+        `job_fields` — поля `recognition_job` поверх фабрики (`reply_deadline`, `published_at`, `metrics`, …);
+        `original_name=None` — файл без имени (запрос не собирается по контракту). Возвращает
+        `(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)
+            file, job = await make_stored_file(
+                session,
+                entity=entity,
+                status=file_status,
+                job_status=job_status,
+                original_name=original_name or "file",
+            )
+            file.original_name = original_name
             for name, value in job_fields.items():
                 setattr(job, name, value)
             await session.flush()

api/tests/test_recognition_exchange.py

@@ -297,17 +297,23 @@ async def test_first_publication_msg_id_follows_attempt(
 
 
 @pytest.mark.integration
+@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
 async def test_invalid_request_marks_job_error_and_pass_continues(
     session_factory: async_sessionmaker[AsyncSession],
     jobs: CommittedJobs,
     nats: NatsConnection,
     subscriber: RequestSubscriber,
+    current: bool,
 ):
-    """Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше."""
-    broken = await jobs.make_queued_job(original_name=None)
+    """Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше; файл
+    текущего задания — `ERROR` (финальная ошибка задания, spec primary-files), файл чужого задания не меняется."""
+    clock = Clock()
+    file_id, broken = await jobs.make_file_job(original_name=None)
+    # новое задание — `processing`, чтобы публикатор его не трогал (в проходе только старое задание файла)
+    new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
     good = await jobs.make_queued_job()
 
-    report = await _publish(session_factory, nats)
+    report = await _publish(session_factory, nats, clock)
 
     assert (report.invalid, report.published) == (1, 1)
     await subscriber.wait_for(good, 1)
@@ -319,6 +325,11 @@ async def test_invalid_request_marks_job_error_and_pass_continues(
     assert job.error_message and "original_name" in job.error_message
     assert job.finished_at is not None
     assert job.request_msg_id is None
+    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)
 
 
 # --- повторы без ack и dlq ---------------------------------------------------------------------------------------