feat(#190): факт технического отказа (TIMEOUT/NO_ACK/SERVICE_ERROR/MAPPING, §14.5)

Задача 10: quality_failure_fact на 6 точках отказа — вахтёр сроков
(expire_deadlines, TIMEOUT, батчем на все просроченные задания), dlq
(_process_one, NO_ACK), невалидный запрос и результат сервиса со
status=error (_process_one/_apply_result, SERVICE_ERROR), неповторяемая
и исчерпанная ошибки маппинга (_fail/_register_failure, MAPPING).
Измерения ЮЛ/ГК/is_scan — отдельным запросом по file_id (батчем у
вахтёра), отсутствие строки ЮЛ не прерывает обработку, только лог.

Восстановление после обрыва сессии: имплементер задачи 10 (agent
a7ad86adedbac1d8a) закончил работу, но коммит и отчёт не состоялись —
диф остался в рабочем дереве незакоммиченным. Коммит и прогон
проверок — контроллером (ruff/mypy/pytest -m "not integration" чисты;
один magic-value в новом тесте поправлен тут же), не ревью находка.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

api/src/services/recognition_exchange.py

@@ -79,8 +79,10 @@ from src.contract.versions import parse_message
 from src.core.config import Settings
 from src.core.nats import NatsPublishError
 from src.core.storage import ObjectStorage
-from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole
+from src.models.directories import CompanyGroup, LegalEntity
+from src.models.enums import FileStatus, QualityFailureKind, RecognitionPipeline, RecognitionStatus, S3BucketRole
 from src.models.files import RecognitionJob, S3Object, StoredFile
+from src.repositories.quality_facts import insert_failure_fact
 from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs
 from src.services.result_mapping import run_mapping_loop
 
@@ -208,6 +210,121 @@ def _short(text: str) -> str:
     return text if len(text) <= ERROR_MESSAGE_MAX_LENGTH else text[: ERROR_MESSAGE_MAX_LENGTH - 1] + "…"
 
 
+# --- факт отказа (quality_failure_fact, §14.5, design quality-metrics §4.3, AURORA-1759) -----------------------------
+
+
+@dataclass(frozen=True)
+class _FailureDimensions:
+    """ЮЛ/ГК и признак скана файла на момент отказа — измерения `quality_failure_fact`, не хранимые в `recognition_job`
+    самом (design §4: факт без FK, имена и id ЮЛ/ГК — копия на момент отказа)."""
+
+    legal_entity_id: uuid.UUID
+    legal_entity_name: str
+    group_id: uuid.UUID | None
+    group_name: str | None
+    is_scan: bool | None
+
+
+async def _failure_dimensions_batch(
+    session: AsyncSession, file_ids: Sequence[uuid.UUID]
+) -> dict[uuid.UUID, _FailureDimensions]:
+    """Измерения ЮЛ/ГК для набора `file_id` одним запросом (вахтёр переводит в `timeout` пачку заданий за проход,
+    §14.5) — без N+1 на задание. Файл без строки ЮЛ (не должно быть при штатной работе — ЮЛ не удаляются) просто
+    не попадает в результат."""
+    if not file_ids:
+        return {}
+    stmt = (
+        select(
+            StoredFile.id,
+            StoredFile.is_scan,
+            LegalEntity.id,
+            LegalEntity.name,
+            LegalEntity.company_group_id,
+            CompanyGroup.name,
+        )
+        .join(LegalEntity, LegalEntity.id == StoredFile.legal_entity_id)
+        .outerjoin(CompanyGroup, CompanyGroup.id == LegalEntity.company_group_id)
+        .where(StoredFile.id.in_(list(file_ids)))
+    )
+    rows = await session.execute(stmt)
+    return {
+        file_id: _FailureDimensions(
+            legal_entity_id=legal_entity_id, legal_entity_name=legal_entity_name, group_id=group_id,
+            group_name=group_name, is_scan=is_scan,
+        )
+        for file_id, is_scan, legal_entity_id, legal_entity_name, group_id, group_name in rows
+    }
+
+
+async def _failure_dimensions(session: AsyncSession, file_id: uuid.UUID) -> _FailureDimensions | None:
+    """Измерения ЮЛ/ГК одного файла (dlq / ошибка валидации запроса / ошибка сервиса — задание обрабатывается по
+    одному); тот же запрос, что у `_failure_dimensions_batch`, набор из одного элемента."""
+    return (await _failure_dimensions_batch(session, [file_id])).get(file_id)
+
+
+async def _record_failure_fact(  # noqa: PLR0913 - все измерения факта отказа именованные
+    session: AsyncSession,
+    *,
+    job_id: uuid.UUID,
+    file_id: uuid.UUID,
+    attempt: int,
+    pipeline: RecognitionPipeline,
+    dims: _FailureDimensions,
+    failure_kind: QualityFailureKind,
+    error_code: str | None,
+    service_name: str | None,
+    failed_at: datetime,
+) -> None:
+    """Тонкая обёртка `insert_failure_fact` (Task 4): измерения ЮЛ/ГК — из уже найденного `dims` (без повторного
+    запроса на каждый вызов у вызывающих с батчем)."""
+    await insert_failure_fact(
+        session,
+        job_id=job_id,
+        file_id=file_id,
+        attempt=attempt,
+        legal_entity_id=dims.legal_entity_id,
+        legal_entity_name=dims.legal_entity_name,
+        group_id=dims.group_id,
+        group_name=dims.group_name,
+        pipeline=pipeline,
+        is_scan=dims.is_scan,
+        service_name=service_name,
+        failure_kind=failure_kind,
+        error_code=error_code,
+        failed_at=failed_at,
+    )
+
+
+async def _write_timeout_failure_facts(session: AsyncSession, rows: Sequence[Any], *, failed_at: datetime) -> None:
+    """Факт отказа `TIMEOUT` на каждое задание, переведённое в `timeout` этим проходом вахтёра (§14.5): задание ещё
+    не получало ответа сервиса — `service_name = NULL`. Пишется для ВСЕХ строк `rows`, включая задания не текущие
+    для своего файла (у файла уже новое задание) — отказ конкретной попытки, а не статус файла."""
+    if not rows:
+        return
+    dimensions = await _failure_dimensions_batch(session, sorted({row.file_id for row in rows}))
+    for row in rows:
+        dims = dimensions.get(row.file_id)
+        if dims is None:
+            logger.warning(
+                "Факт отказа TIMEOUT не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}",
+                job_id=row.id,
+                file_id=row.file_id,
+            )
+            continue
+        await _record_failure_fact(
+            session,
+            job_id=row.id,
+            file_id=row.file_id,
+            attempt=row.attempt,
+            pipeline=row.pipeline,
+            dims=dims,
+            failure_kind=QualityFailureKind.TIMEOUT,
+            error_code=ERROR_TIMEOUT,
+            service_name=None,
+            failed_at=failed_at,
+        )
+
+
 async def _process_one(  # noqa: PLR0913 - один шаг прохода: все параметры прохода
     session: AsyncSession,
     nats: RequestPublisher,
@@ -239,6 +356,27 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
             )
         )
         await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
+        # §14.5: факт отказа NO_ACK — задание так и не получило ответа сервиса, service_name неизвестен
+        dims = await _failure_dimensions(session, row.file_id)
+        if dims is None:
+            logger.warning(
+                "Факт отказа NO_ACK не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}",
+                job_id=row.id,
+                file_id=row.file_id,
+            )
+        else:
+            await _record_failure_fact(
+                session,
+                job_id=row.id,
+                file_id=row.file_id,
+                attempt=row.attempt,
+                pipeline=row.pipeline,
+                dims=dims,
+                failure_kind=QualityFailureKind.NO_ACK,
+                error_code=ERROR_NO_ACK,
+                service_name=None,
+                failed_at=now,
+            )
         report.dlq += 1
         logger.warning(
             "Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, "
@@ -263,6 +401,28 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
         )
         # финальная ошибка задания — файл текущего задания ERROR (spec primary-files «Жизненный цикл статуса файла»)
         await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
+        # §14.5: любой переход recognition_job.status -> error — SERVICE_ERROR, независимо от причины (наша
+        # валидация запроса или ошибка, о которой сообщил сервис, design Review Focus основного плана)
+        dims = await _failure_dimensions(session, row.file_id)
+        if dims is None:
+            logger.warning(
+                "Факт отказа SERVICE_ERROR не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}",
+                job_id=row.id,
+                file_id=row.file_id,
+            )
+        else:
+            await _record_failure_fact(
+                session,
+                job_id=row.id,
+                file_id=row.file_id,
+                attempt=row.attempt,
+                pipeline=row.pipeline,
+                dims=dims,
+                failure_kind=QualityFailureKind.SERVICE_ERROR,
+                error_code=ERROR_INVALID_REQUEST,
+                service_name=None,
+                failed_at=now,
+            )
         report.invalid += 1
         logger.error(
             "Запрос распознавания не собран по контракту, задание переведено в error. "
@@ -373,7 +533,8 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *,
     """Вахтёр сроков (spec «Таймаут ожидания результата»): `processing`/`published` с `reply_deadline < now` →
     `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE … RETURNING id` с условием по
     статусу: результат, успевший перевести задание в финальный статус, не перетирается. В той же транзакции файлы этих
-    заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`. Возвращает число заданий."""
+    заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`, и для КАЖДОГО (а не только
+    текущего) — факт отказа `TIMEOUT` (§14.5). Возвращает число заданий."""
     moment = (now or _utcnow)()
     stmt = (
         update(RecognitionJob)
@@ -387,12 +548,14 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *,
             error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)",
             finished_at=moment,
         )
-        .returning(RecognitionJob.id)
+        .returning(RecognitionJob.id, RecognitionJob.file_id, RecognitionJob.pipeline, RecognitionJob.attempt)
         .execution_options(synchronize_session=False)
     )
     async with session_factory() as session, session.begin():
-        job_ids = list((await session.execute(stmt)).scalars())
+        rows = (await session.execute(stmt)).all()
+        job_ids = [row.id for row in rows]
         files = await set_file_status_for_jobs(session, job_ids, FileStatus.ERROR, now=moment)
+        await _write_timeout_failure_facts(session, rows, failed_at=moment)
     count = len(job_ids)
     if count:
         logger.warning(
@@ -726,6 +889,29 @@ async def _apply_result(
     job.result_s3_object_id = result_s3_object_id
     job.result_msg_id = msg_id
     file_changed = await _apply_result_file_status(session, job, now)
+    if job.status is RecognitionStatus.error:
+        # §14.5: любой переход recognition_job.status -> error — SERVICE_ERROR; service_name/error_code уже
+        # заполнены выше из сообщения сервиса — здесь единственная точка отказа, где они известны
+        dims = await _failure_dimensions(session, job.file_id)
+        if dims is None:
+            logger.warning(
+                "Факт отказа SERVICE_ERROR не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}",
+                job_id=job.id,
+                file_id=job.file_id,
+            )
+        else:
+            await _record_failure_fact(
+                session,
+                job_id=job.id,
+                file_id=job.file_id,
+                attempt=job.attempt,
+                pipeline=job.pipeline,
+                dims=dims,
+                failure_kind=QualityFailureKind.SERVICE_ERROR,
+                error_code=job.error_code,
+                service_name=job.service_name,
+                failed_at=now,
+            )
     logger.info(
         "Результат распознавания принят. job_id={job_id}, status={status}, service={service}, msg_id={msg_id}, "
         "error_code={error_code}, file_changed={file_changed}",

api/src/services/result_mapping.py

@@ -73,6 +73,7 @@ from src.models.enums import (
     OsvDataType,
     OsvGroupType,
     PeriodQuarter,
+    QualityFailureKind,
     QualityOutcome,
     RecognitionStatus,
     StatementLineStatus,
@@ -82,7 +83,7 @@ from src.models.osv import OsvEntry
 from src.models.statements import DocumentIssue
 from src.repositories import statements as statements_repo
 from src.repositories.counterparties import CounterpartyIndex
-from src.repositories.quality_facts import insert_intake_fact, insert_outcome_fact
+from src.repositories.quality_facts import insert_failure_fact, insert_intake_fact, insert_outcome_fact
 from src.services import consolidation_documents
 from src.services.audit import AuditService
 from src.services.file_status import set_file_status_for_job
@@ -129,6 +130,10 @@ INTAKE_FACT_DOCUMENT_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOM
 STORED_FILE_OBJECT_TYPE = "stored_file"  # `audit_event.object_type` события FILE_REPLACE
 SUPERSEDE_LOCK_PREFIX = "supersede"  # пространство ключей advisory-блокировки автозамены в пределах ЮЛ
 _FORBIDDEN_KEY_SEGMENTS = frozenset({"", ".", ".."})
+# §14.5 (design quality-metrics §4.3): единый короткий error_code факта отказа MAPPING — что именно не так (схема,
+# S3, объект не найден, БД) остаётся в `metrics.mapping_error`/логе (M-1), у `quality_failure_fact` нет текстовой
+# колонки для детали; свой код, не пересекается с ERROR_TIMEOUT/ERROR_NO_ACK/ERROR_INVALID_REQUEST обмена
+MAPPING_FAILURE_ERROR_CODE = "MAPPING_FAILED"
 
 Clock = Callable[[], datetime]
 SleepFn = Callable[[float], Awaitable[None]]
@@ -1074,17 +1079,86 @@ async def _lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile |
     return (await session.execute(stmt)).scalar_one_or_none()
 
 
+# --- факт отказа (quality_failure_fact, §14.5, design quality-metrics §4.3, AURORA-1759) -----------------------------
+
+
+@dataclass(frozen=True)
+class _FailureDimensions:
+    """ЮЛ/ГК и признак скана файла на момент отказа маппинга — измерения `quality_failure_fact` (design §4: факт без
+    FK, имена и id ЮЛ/ГК — копия на момент отказа)."""
+
+    legal_entity_id: uuid.UUID
+    legal_entity_name: str
+    group_id: uuid.UUID | None
+    group_name: str | None
+    is_scan: bool | None
+
+
+async def _failure_dimensions(session: AsyncSession, file_id: uuid.UUID) -> _FailureDimensions | None:
+    """ЮЛ/ГК и признак скана файла по `file_id` (`session.get`, тот же приём, что у `_write_intake_facts`): файл или
+    ЮЛ не найдены (не должно быть при штатной работе) — `None`, факт отказа не пишется, остальная обработка не
+    прерывается."""
+    file = await session.get(StoredFile, file_id)
+    if file is None:
+        return None
+    owner = await session.get(LegalEntity, file.legal_entity_id)
+    if owner is None:
+        return None
+    group_name: str | None = None
+    if owner.company_group_id is not None:
+        group = await session.get(CompanyGroup, owner.company_group_id)
+        group_name = group.name if group is not None else None
+    return _FailureDimensions(
+        legal_entity_id=owner.id,
+        legal_entity_name=owner.name,
+        group_id=owner.company_group_id,
+        group_name=group_name,
+        is_scan=file.is_scan,
+    )
+
+
+async def _write_mapping_failure_fact(session: AsyncSession, job: RecognitionJob, *, failed_at: datetime) -> None:
+    """Факт отказа `MAPPING` (§14.5) — `job` уже дошёл до `done`/`needs_check` (`service_name`/`attempt` сервиса
+    известны, в отличие от отказов обмена задачи 10 sites 1-4), а маппинг результата не удался."""
+    dims = await _failure_dimensions(session, job.file_id)
+    if dims is None:
+        logger.warning(
+            "Факт отказа MAPPING не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}",
+            job_id=job.id,
+            file_id=job.file_id,
+        )
+        return
+    await insert_failure_fact(
+        session,
+        job_id=job.id,
+        file_id=job.file_id,
+        attempt=job.attempt,
+        legal_entity_id=dims.legal_entity_id,
+        legal_entity_name=dims.legal_entity_name,
+        group_id=dims.group_id,
+        group_name=dims.group_name,
+        pipeline=job.pipeline,
+        is_scan=dims.is_scan,
+        service_name=job.service_name,
+        failure_kind=QualityFailureKind.MAPPING,
+        error_code=MAPPING_FAILURE_ERROR_CODE,
+        failed_at=failed_at,
+    )
+
+
 async def _fail(
     session: AsyncSession, candidate: Candidate, message: str, now: datetime, *, cause: BaseException | None = None
 ) -> None:
     """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего).
-    `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1)."""
+    `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1).
+    §14.5: факт отказа `MAPPING`."""
     job = await _lock_job(session, candidate.job_id)
     if job is None:
         return
     job.metrics = {**_metrics(job), MAPPING_ERROR_KEY: _short(message)}
     await session.flush()
     changed = await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
+    await _write_mapping_failure_fact(session, job, failed_at=now)
     logger.opt(exception=cause).error(
         "Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, "
         "file_id={file_id}, file_changed={changed}, error={error}",
@@ -1103,8 +1177,9 @@ async def _register_failure(
     max_attempts: int,
     now: datetime,
 ) -> bool:
-    """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`.
+    """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR` и
-    Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул очередь)."""
+    факт отказа `MAPPING` (§14.5). Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул
+    очередь)."""
     async with session_factory() as session, session.begin():
         job = await _lock_job(session, candidate.job_id)
         if job is None:
@@ -1122,6 +1197,7 @@ async def _register_failure(
         await session.flush()
         if exhausted:
             await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
+            await _write_mapping_failure_fact(session, job, failed_at=now)
     logger.opt(exception=exc).error(
         "Ошибка маппинга результата, транзакция откачена. error_category=mapping_error, job_id={job_id}, "
         "file_id={file_id}, attempts={attempts}, file_error={exhausted}",

api/tests/test_recognition_exchange.py

@@ -23,17 +23,22 @@ import pytest_asyncio
 from jsonschema import Draft202012Validator, FormatChecker
 from nats.aio.client import Client as NATS
 from nats.aio.msg import Msg
-from sqlalchemy import JSON, update
+from sqlalchemy import JSON, select, update
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 
 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 FileStatus, FileType, RecognitionPipeline, RecognitionStatus
+from src.models.directories import LegalEntity
-from src.models.files import RecognitionJob
+from src.models.enums import FileStatus, FileType, QualityFailureKind, RecognitionPipeline, RecognitionStatus
+from src.models.files import RecognitionJob, StoredFile
+from src.models.quality_facts import QualityFailureFact
 from src.services import recognition_exchange as rx
 from src.services.recognition_exchange import (
+    ERROR_INVALID_REQUEST,
+    ERROR_NO_ACK,
+    ERROR_TIMEOUT,
     WATCHDOG_PERIOD_SECONDS,
     ExchangeTasks,
     expire_deadlines,
@@ -41,6 +46,7 @@ from src.services.recognition_exchange import (
     run_publisher_loop,
     run_watchdog_loop,
 )
+from src.services.result_mapping import MAPPING_FAILURE_ERROR_CODE
 from tests.factories import CommittedJobs
 
 CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts"
@@ -187,6 +193,25 @@ def _publish(
     return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now)
 
 
+async def _legal_entity(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> LegalEntity:
+    """ЮЛ-владелец файла (для сравнения с измерениями `quality_failure_fact`, §14.5)."""
+    async with session_factory() as session:
+        file = await session.get(StoredFile, file_id)
+        assert file is not None
+        entity = await session.get(LegalEntity, file.legal_entity_id)
+        assert entity is not None
+        return entity
+
+
+async def _failure_fact(
+    session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID
+) -> QualityFailureFact | None:
+    """Факт отказа задания `job_id`, если записан (`job_id` — unique, не больше одного)."""
+    async with session_factory() as session:
+        stmt = select(QualityFailureFact).where(QualityFailureFact.job_id == job_id)
+        return (await session.execute(stmt)).scalar_one_or_none()
+
+
 # --- юнит: соответствие enum, константы ------------------------------------------------------------------------
 
 
@@ -200,6 +225,13 @@ def test_orm_pipeline_matches_request_subjects():
     assert {m.value for m in RecognitionPipeline} == {"tabular", "image"}
 
 
+def test_mapping_failure_error_code_is_distinct_from_exchange_codes():
+    """§14.5: `error_code` отказа MAPPING (`result_mapping.py`, задача 10) — свой короткий код, не пересекается с
+    кодами обмена (ERROR_TIMEOUT/ERROR_NO_ACK/ERROR_INVALID_REQUEST) — иначе по `error_code` без `failure_kind`
+    нельзя было бы отличить отказ обмена от отказа маппинга. Чисто логическая проверка, без БД."""
+    assert MAPPING_FAILURE_ERROR_CODE not in {ERROR_TIMEOUT, ERROR_NO_ACK, ERROR_INVALID_REQUEST}
+
+
 def test_watchdog_period_is_30_seconds():
     assert WATCHDOG_PERIOD_SECONDS == 30  # noqa: PLR2004 - значение из spec
 
@@ -395,6 +427,41 @@ async def test_invalid_request_marks_job_error_and_pass_continues(
         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_invalid_request_writes_service_error_failure_fact(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool
+):
+    """§14.5: запрос, не собранный по контракту, — тоже `SERVICE_ERROR` (любой переход `recognition_job.status ->
+    error` — SERVICE_ERROR, design Review Focus); факт пишется независимо от того, текущее это задание файла
+    (как и у TIMEOUT вахтёра) — отказ конкретной попытки. `nats.publish` до этой ветки не доходит — фейковый NATS
+    без реального соединения достаточно."""
+    clock = Clock()
+    file_id, broken = await jobs.make_file_job(original_name=None)
+    if not current:
+        await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
+    job = await jobs.get(broken)
+    entity = await _legal_entity(session_factory, file_id)
+
+    report = await _publish(session_factory, FakeNats(), clock)
+
+    assert report.invalid == 1
+    fact = await _failure_fact(session_factory, broken)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job.attempt
+    assert fact.legal_entity_id == entity.id
+    assert fact.legal_entity_name == entity.name
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline is RecognitionPipeline.tabular
+    assert fact.is_scan is None
+    assert fact.service_name is None
+    assert fact.failure_kind is QualityFailureKind.SERVICE_ERROR
+    assert fact.error_code == ERROR_INVALID_REQUEST
+    assert fact.failed_at == clock()
+
+
 # --- повторы без ack и dlq ---------------------------------------------------------------------------------------
 
 
@@ -689,6 +756,62 @@ async def test_expire_deadlines_moves_current_file_to_error(
     assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
 
 
+@pytest.mark.integration
+async def test_expire_deadlines_writes_timeout_failure_fact(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
+):
+    """§14.5: вахтёр переводит задание в `timeout` — факт отказа TIMEOUT; сервис так и не ответил — `service_name`
+    пуст. `.returning()` `expire_deadlines` отдаёт `file_id`/`pipeline`/`attempt` тем же `UPDATE`, без второго
+    запроса на задание."""
+    clock = Clock()
+    file_id, job_id = await jobs.make_file_job(
+        job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1)
+    )
+    job = await jobs.get(job_id)
+    entity = await _legal_entity(session_factory, file_id)
+
+    assert await expire_deadlines(session_factory, now=clock) == 1
+
+    fact = await _failure_fact(session_factory, job_id)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job.attempt
+    assert fact.legal_entity_id == entity.id
+    assert fact.legal_entity_name == entity.name
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline is RecognitionPipeline.tabular
+    assert fact.is_scan is None
+    assert fact.service_name is None
+    assert fact.failure_kind is QualityFailureKind.TIMEOUT
+    assert fact.error_code == ERROR_TIMEOUT
+    assert fact.failed_at == clock()
+
+
+@pytest.mark.integration
+async def test_expire_deadlines_writes_failure_fact_for_non_current_job(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
+):
+    """§14.5: факт отказа пишется на КАЖДОЕ просроченное задание, а не только на текущее задание файла (design
+    §14.5: «один файл может быть и в отказах, и в исходах»; 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),
+    )
+    await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)
+
+    assert await expire_deadlines(session_factory, now=clock) == 1
+
+    fact = await _failure_fact(session_factory, old_job)
+    assert fact is not None
+    assert fact.failure_kind is QualityFailureKind.TIMEOUT
+    file = await jobs.file(file_id)
+    assert file.status is FileStatus.UPLOADED  # файл не трогается (текущее задание — другое)
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
 async def test_expire_deadlines_keeps_processed_and_superseded_file(
@@ -754,6 +877,40 @@ async def test_dlq_moves_current_file_to_error(
         assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
 
 
+@pytest.mark.integration
+async def test_dlq_writes_no_ack_failure_fact(session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs):
+    """§14.5: повторы без ack исчерпаны (`dlq`) — факт отказа NO_ACK; задание не получило ответа сервиса —
+    `service_name` пуст."""
+    clock = Clock()
+    file_id, job_id = await jobs.make_file_job(
+        job_status=RecognitionStatus.published,
+        published_at=clock() - ACK * 2,
+        request_msg_id="dlq-fact-test:1",
+        reply_deadline=clock() + timedelta(hours=1),
+        metrics={"republish_count": MAX_REPUBLISH},
+    )
+    job = await jobs.get(job_id)
+    entity = await _legal_entity(session_factory, file_id)
+
+    report = await _publish(session_factory, FakeNats(), clock)
+
+    assert report.dlq == 1
+    fact = await _failure_fact(session_factory, job_id)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job.attempt
+    assert fact.legal_entity_id == entity.id
+    assert fact.legal_entity_name == entity.name
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline is RecognitionPipeline.tabular
+    assert fact.is_scan is None
+    assert fact.service_name is None
+    assert fact.failure_kind is QualityFailureKind.NO_ACK
+    assert fact.error_code == ERROR_NO_ACK
+    assert fact.failed_at == clock()
+
+
 # --- фоновые циклы -----------------------------------------------------------------------------------------------
 
 

api/tests/test_recognition_inbox.py

@@ -35,8 +35,10 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 from src.contract import NATS_MSG_ID_HEADER, ack_msg_id, result_msg_id
 from src.core.config import get_settings
 from src.core.nats import NatsConnection
-from src.models.enums import FileStatus, RecognitionStatus, S3BucketRole
+from src.models.directories import LegalEntity
+from src.models.enums import FileStatus, QualityFailureKind, RecognitionStatus, S3BucketRole
 from src.models.files import RecognitionJob, S3Object, StoredFile
+from src.models.quality_facts import QualityFailureFact
 from src.services import recognition_exchange as rx
 from src.services.recognition_exchange import (
     FINAL_STATUSES,
@@ -176,6 +178,25 @@ async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID:
     return (await jobs.get(job_id)).file_id
 
 
+async def _legal_entity(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> LegalEntity:
+    """ЮЛ-владелец файла (для сравнения с измерениями `quality_failure_fact`, §14.5)."""
+    async with session_factory() as session:
+        file = await session.get(StoredFile, file_id)
+        assert file is not None
+        entity = await session.get(LegalEntity, file.legal_entity_id)
+        assert entity is not None
+        return entity
+
+
+async def _failure_fact(
+    session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID
+) -> QualityFailureFact | None:
+    """Факт отказа задания `job_id`, если записан (`job_id` — unique, не больше одного)."""
+    async with session_factory() as session:
+        stmt = select(QualityFailureFact).where(QualityFailureFact.job_id == job_id)
+        return (await session.execute(stmt)).scalar_one_or_none()
+
+
 async def _handle(session_factory: async_sessionmaker[AsyncSession], msg: Any, **kwargs: Any) -> None:
     await handle_message(msg, session_factory=session_factory, **kwargs)
 
@@ -535,6 +556,58 @@ async def test_service_error_without_result_s3(session_factory, jobs: CommittedJ
     assert await _s3_total(session_factory) == total_before
 
 
+@pytest.mark.integration
+async def test_service_error_result_writes_failure_fact(session_factory, jobs: CommittedJobs):
+    """§14.5 (задача 10, site 4): результат сервиса со `status=error` — факт отказа SERVICE_ERROR; в отличие от
+    отказов обмена без ответа (TIMEOUT/NO_ACK/INVALID_REQUEST — задачи 10 sites 1-3), здесь сервис уже ответил —
+    `service_name`/`error_code` берутся из уже заполненных полей задания, а не оставляются пустыми."""
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    entity = await _legal_entity(session_factory, file_id)
+    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)
+    fact = await _failure_fact(session_factory, job_id)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job["attempt"]
+    assert fact.legal_entity_id == entity.id
+    assert fact.legal_entity_name == entity.name
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline == job["pipeline"]
+    assert fact.is_scan is None
+    assert fact.service_name == "tabular-parser"
+    assert fact.failure_kind is QualityFailureKind.SERVICE_ERROR
+    assert fact.error_code == "NOT_IMPLEMENTED"
+    assert fact.failed_at == job["finished_at"]
+
+
+@pytest.mark.integration
+async def test_result_for_final_status_before_error_does_not_write_second_failure_fact(
+    session_factory, jobs: CommittedJobs
+):
+    """Дубликат/запоздавший результат для задания, уже `error` (R-22 — без изменений): повторного факта нет —
+    `job_id` в `quality_failure_fact` unique, да и `_apply_result` для финального статуса возвращает `False` раньше
+    любых изменений (Review Focus)."""
+    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
+    file_id = await _file_id(jobs, job_id)
+    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)))
+    first = await _failure_fact(session_factory, job_id)
+    assert first is not None
+
+    # повтор того же результата (дубликат по result_msg_id) — задание уже в финальном статусе, без изменений
+    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)))
+
+    async with session_factory() as session:
+        stmt = select(func.count()).select_from(QualityFailureFact).where(QualityFailureFact.job_id == job_id)
+        count = (await session.execute(stmt)).scalar_one()
+    assert count == 1
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize(
     "status",

api/tests/test_result_mapping.py

@@ -50,12 +50,13 @@ from src.models.enums import (
     OsvDataType,
     OsvGroupType,
     PeriodQuarter,
+    QualityFailureKind,
     QualityOutcome,
     RecognitionStatus,
 )
 from src.models.files import Document, RecognitionJob, StoredFile
 from src.models.osv import OsvEntry
-from src.models.quality_facts import QualityIntakeFact, QualityOutcomeFact
+from src.models.quality_facts import QualityFailureFact, QualityIntakeFact, QualityOutcomeFact
 from src.models.statements import (
     DocumentIssue,
     FinancialStatementLine,
@@ -2317,4 +2318,141 @@ def test_intake_fact_document_types_excludes_disclosure_and_other() -> None:
     чистая проверка множества типов хука, без БД (в отличие от прочих тестов `_map_candidate` в этом файле)."""
     assert rm.INTAKE_FACT_DOCUMENT_TYPES == {DocumentType.BALANCE, DocumentType.INCOME, DocumentType.OSV}
     assert DocumentType.DISCLOSURE not in rm.INTAKE_FACT_DOCUMENT_TYPES
+
+
+# --- факт отказа маппинга (quality_failure_fact, §14.5, design quality-metrics §4.3, задача 10 sites 5-6) ----------
+
+
+async def _failure_fact(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> QualityFailureFact | None:
+    async with maker() as session:
+        return (
+            await session.execute(select(QualityFailureFact).where(QualityFailureFact.job_id == job_id))
+        ).scalar_one_or_none()
+
+
+async def test_mapping_fatal_writes_mapping_failure_fact(committed_sessionmaker: async_sessionmaker[AsyncSession]):
+    """Задача 10, site 5 (§14.5): неповторяемая ошибка маппинга (`MappingFatal` — здесь ключ результата вне папки
+    задания, как в `test_result_key_outside_job_folder_is_error`) — факт отказа `MAPPING`. `job` в `_fail` уже
+    `done`/`needs_check` — `service_name` берётся из задания, а не оставляется пустым (в отличие от отказов обмена
+    задачи 10 sites 1-3, где сервис ещё не ответил)."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    file_id, job_id, entity_id = await _prepare(
+        maker, storage, _minimal(), key=f"results/{uuid.uuid4()}/{uuid.uuid4()}/result.json"
+    )
+    job_before = await _job(maker, job_id)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    fact = await _failure_fact(maker, job_id)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job_before.attempt
+    assert fact.legal_entity_id == entity_id
+    assert fact.legal_entity_name == OWNER_NAME
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline == job_before.pipeline
+    assert fact.is_scan is None
+    assert fact.service_name == job_before.service_name  # `_prepare` не выставляет service_name — оба None
+    assert fact.failure_kind is QualityFailureKind.MAPPING
+    assert fact.error_code == rm.MAPPING_FAILURE_ERROR_CODE
+    assert fact.failed_at is not None
+
+
+async def test_register_failure_writes_mapping_failure_fact_only_when_exhausted(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
+):
+    """Задача 10, site 6 (§14.5): повторяемая ошибка маппинга (как в `test_mapping_failure_retries_then_error`) —
+    факт отказа `MAPPING` пишется только когда попытки исчерпаны (`exhausted`), не на каждом откате транзакции."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    file_id, job_id, entity_id = await _prepare(maker, storage, _minimal())
+
+    async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
+        raise RuntimeError("подставная ошибка БД при вставке строк")
+
+    monkeypatch.setattr(rm, "map_file", failing_map_file)
+
+    for _attempt in (1, 2):
+        assert await process_once(maker, storage, _settings()) is False
+        assert await _failure_fact(maker, job_id) is None  # попытки не исчерпаны — факта ещё нет
+
+    job_before = await _job(maker, job_id)
+    assert await process_once(maker, storage, _settings()) is True  # третья попытка исчерпана — файл ERROR
+
+    fact = await _failure_fact(maker, job_id)
+    assert fact is not None
+    assert fact.file_id == file_id
+    assert fact.attempt == job_before.attempt
+    assert fact.legal_entity_id == entity_id
+    assert fact.legal_entity_name == OWNER_NAME
+    assert fact.group_id is None
+    assert fact.group_name is None
+    assert fact.pipeline == job_before.pipeline
+    assert fact.is_scan is None
+    assert fact.service_name is None
+    assert fact.failure_kind is QualityFailureKind.MAPPING
+    assert fact.error_code == rm.MAPPING_FAILURE_ERROR_CODE
+    assert fact.failed_at is not None
+
+
+async def test_file_reprocess_keeps_failure_fact_of_first_attempt_and_adds_new_intake(
+    committed_sessionmaker: async_sessionmaker[AsyncSession],
+):
+    """Шаг 7 плана (§14.5): `FILE_REPROCESS` (новое `recognition_job` с `attempt + 1` на тот же файл —
+    `src/services/file_retry.py` `FileRetryService._new`, здесь воспроизведено напрямую вставкой задания — тем же
+    приёмом, что `test_result_of_non_current_job_is_ignored`/`test_job_replaced_while_reading_result` выше) не
+    теряет факт отказа первой попытки: он остаётся на `job_id` первого задания (`attempt=1`), а успешная вторая
+    попытка пишет свой `quality_intake_fact` (`attempt=2`, задача 5) — оба факта сосуществуют на одном `file_id`,
+    это не двойной учёт одного события (design §14.5: «один файл может быть и в отказах, и в исходах»)."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    async with maker() as session, session.begin():
+        entity = await make_legal_entity(session, name=OWNER_NAME)
+        file, job1 = await make_stored_file(
+            session, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
+        )
+        job1.reply_deadline = _now() - timedelta(minutes=1)
+        file_id, job1_id = file.id, job1.id
+
+    # первая попытка: вахтёр переводит задание в timeout -> файл ERROR -> факт отказа TIMEOUT (attempt=1, задача 10
+    # sites 1-4, уже на этой ветке)
+    assert await expire_deadlines(maker) == 1
+    assert (await _file(maker, file_id)).status is FileStatus.ERROR
+
+    # FILE_REPROCESS: новое задание attempt=2, файл снова ждёт результата (как `FileRetryService._new`)
+    async with maker() as session, session.begin():
+        new_file = await session.get(StoredFile, file_id)
+        job1_fresh = await session.get(RecognitionJob, job1_id)
+        assert new_file is not None and job1_fresh is not None
+        job2 = RecognitionJob(
+            file_id=file_id,
+            pipeline=job1_fresh.pipeline,
+            request_subject=job1_fresh.request_subject,
+            schema_version="1.0",
+            status=RecognitionStatus.done,
+            attempt=2,
+            finished_at=_now(),
+        )
+        session.add(job2)
+        await session.flush()
+        new_file.current_recognition_job_id = job2.id
+        new_file.status = FileStatus.PROCESSING
+        job2_id = job2.id
+        payload = _minimal() | {"job_id": str(job2.id), "file_id": str(file_id)}
+        await make_result_object(session, job=job2, storage=storage, result=payload)
+
+    # вторая попытка — успех: маппинг создаёт документ и пишет свой факт приёма (attempt=2)
+    assert await process_once(maker, storage, _settings()) is True
+
+    async with maker() as session:
+        facts = (await session.execute(select(QualityFailureFact))).scalars().all()
+    assert len(facts) == 1
+    assert facts[0].job_id == job1_id
+    assert facts[0].attempt == 1
+    assert facts[0].file_id == file_id
+
+    [document] = await _documents(maker, file_id)
+    intake = await _intake_fact(maker, document.id)
+    assert intake.job_id == job2_id
+    assert intake.attempt == 2  # noqa: PLR2004 - вторая попытка (job2) этого файла
+    assert intake.file_id == file_id
     assert DocumentType.OTHER not in rm.INTAKE_FACT_DOCUMENT_TYPES