NATS: subject'ы как в старом АРМ — files.*_uploaded, files.tabular_parsed (#22) #23

Merged
aep merged 1 commits from feat/22-nats-files-subjects into main 2026-10-08 22:48:32 +03:00
20 changed files with 164 additions and 39 deletions

View File

@ -8,7 +8,7 @@
публикуется **без замены**), забрать **все** `Исходящие/<job_id>/*.json` (ИС кладёт по одному json на каждый
распознанный документ), склеить и сохранить сырой ответ в S3 `raw/{job_id}.json` (в БД не регистрируется), очистить
горячую папку (только при `SCAN_CLEANUP=true`), нормализовать в `../docs/contracts/recognition-result.schema.json`
(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `recognition.result`
(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `files.tabular_parsed`
(`done` / `needs_check` / `error`). Таймаут → ошибка `TIMEOUT` по `min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS=300)`,
старт — запись файла во «Входящие». **Нет доступа к БД.**

View File

@ -2,7 +2,7 @@
Адаптер внешней ИС Сканирование (Content AI) для АРМ КА. Принимает из NATS запросы на распознавание изображений
(PDF/PNG/JPG), передаёт файл в ИС через «горячие папки», забирает ответ, сохраняет сырой ответ в S3, нормализует его
и публикует результат в `recognition.result`. Контракт обмена — `../docs/contracts/` (v1.2: `statementDocument`;
и публикует результат в `files.tabular_parsed`. Контракт обмена — `../docs/contracts/` (v1.2: `statementDocument`;
`schema_version` в конвертах и файле результата сервис по-прежнему пишет `1.0` — `CONTRACT_VERSION`, решение №14
change'а `slice-3a-ocr-normalizer`: поля 1.1–1.2 документы PDF-пути не используют), архитектура —
`../docs/context/03-architecture.md`.
@ -23,9 +23,9 @@ change'а `slice-3a-ocr-normalizer`: поля 1.1–1.2 документы PDF-
## Поток обработки
```
recognition.image.request (NATS, queue group scan-gateway)
files.img_uploaded (NATS, queue group scan-gateway)
1. разбор и проверка запроса ──нет/не по схеме──► error SCHEMA_MISMATCH (без ack)
2. ack: recognition.result, status=processing (до чтения S3 и до ожидания слота)
2. ack: files.tabular_parsed, status=processing (до чтения S3 и до ожидания слота)
3. слот (SCAN_MAX_INFLIGHT одновременных заданий)
4. файл из S3 (s3.bucket/s3.key)
5. горячая папка: <Input>/<job_id>.<ext> (через <Tmp>/.<job_id>.<ext>.part, без замены)
@ -34,7 +34,7 @@ recognition.image.request (NATS, queue group scan-gateway)
7. склейка всех *.json каталога ──► S3 raw/{job_id}.json (до нормализации)
8. очистка горячей папки (только при SCAN_CLEANUP=true)
9. нормализатор (OcrNormalizer) ──► (done/needs_check) S3 results/{file_id}/{job_id}/result.json
10. recognition.result: итоговый конверт (done | needs_check | error), в нём raw_s3_key
10. files.tabular_parsed: итоговый конверт (done | needs_check | error), в нём raw_s3_key
```
Каждое задание заканчивается опубликованным конвертом: сбой превращается в `error` с кодом (таблица ниже), а не в
@ -370,8 +370,8 @@ MinIO и NATS); те же значения подхватываются из `EN
Схемы и примеры — `../docs/contracts/`; их байт-в-байт копии лежат в `api/contracts/` (идентичность проверяет unit-тест
`api/tests/unit/test_contract.py`, входит в `make check`), pydantic-модели — `api/src/contract/`.
- Запрос: subject `recognition.image.request`, подписка в queue group `scan-gateway`, `Nats-Msg-Id = <job_id>:<attempt>`.
- Ответы — subject `recognition.result`: ack `status=processing`, `Nats-Msg-Id = ack:<job_id>:<attempt>`; итог
- Запрос: subject `files.img_uploaded`, подписка в queue group `scan-gateway`, `Nats-Msg-Id = <job_id>:<attempt>`.
- Ответы — subject `files.tabular_parsed`: ack `status=processing`, `Nats-Msg-Id = ack:<job_id>:<attempt>`; итог
`status = done | needs_check | error`, `Nats-Msg-Id = result:<job_id>:<attempt>`. `service.name = "scan-gateway"`.
- Ключи S3: исходник — из запроса (`s3.key`), сырой ответ (склейка файлов ИС) — `raw/{job_id}.json`, результат —
`results/{file_id}/{job_id}/result.json`. Бакет один.
@ -410,8 +410,8 @@ curl -s localhost:8000/healthcheck
PDF из `../fixtures/ocr` (`aep_3.pdf` -> набор `aep_3.json` + `aep_3_ofr.json`, `aep_6.pdf` -> `aep_6.json`,
`aep_5.PDF` -> `aep_5.json`); любое другое содержимое даёт `INVALID_DOCUMENT`, если не задан `SCAN_EMULATOR_DEFAULT`.
`original_name` в запросе определяет только расширение входящего файла.
3. Подписаться на `recognition.result` любым клиентом NATS.
4. Опубликовать в `recognition.image.request` тело по образцу `api/contracts/examples/request-image.json`
3. Подписаться на `files.tabular_parsed` любым клиентом NATS.
4. Опубликовать в `files.img_uploaded` тело по образцу `api/contracts/examples/request-image.json`
(свои `job_id` и `file_id` — UUID, `original_name`, `s3.key`, `reply_deadline` в будущем, `attempt` = 1) с заголовком
`Nats-Msg-Id: <job_id>:1`.
5. Ожидать: ack `processing`, через `SCAN_EMULATOR_DELAY` (+ до `SCAN_SETTLE_SECONDS + SCAN_POLL_SECONDS` после первого опроса, увидевшего набор) —

View File

@ -27,7 +27,7 @@ OPENBAO__NATS__NATS_URL=nats://localhost:4222
OPENBAO__NATS__NATS_SEED=
OPENBAO__NATS__NATS_JWT=
# Включает подписку на recognition.image.request. false — воркер не запускается (сервис поднимется без NATS).
# Включает подписку на files.img_uploaded. false — воркер не запускается (сервис поднимется без NATS).
NATS_ENABLED=true
# PUBLISH_RETRIES — число повторов публикации ack/результата в NATS после первой попытки (при 3 — до 4 попыток всего);
# DONE_CACHE_SIZE — размер LRU обработанных job_id (защита от повторной доставки запроса)

View File

@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-request/v1",
"title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0",
"description": "Тело сообщения на subject recognition.tabular.request или recognition.image.request; subject должен соответствовать полю pipeline (tabular → recognition.tabular.request, image → recognition.image.request). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
"description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object",
"required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"],

View File

@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "arm-ka/recognition-result-message/v1",
"title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0",
"description": "Тело сообщения на subject recognition.result. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
"description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
"$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
"type": "object",
"required": ["schema_version", "job_id", "file_id", "status", "service"],

View File

@ -5,7 +5,7 @@
- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер
(`Worker` с `OcrNormalizer`). Справочники нормализатора загружаются из `SCAN_RULES_DIR` в `create_app`: битый
справочник — `ValueError`, сервис не стартует. Подключение к NATS блокирует до доступности сервера, поэтому идёт
фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после
фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `files.img_uploaded` появляется после
подключения. Сбой этой задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`,
HTTP-сервер жив.
Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3.

View File

@ -1,4 +1,4 @@
"""Конверт сообщения на `recognition.result`: `recognition-result-message.schema.json` v1.0.
"""Конверт сообщения на `files.tabular_parsed`: `recognition-result-message.schema.json` v1.0.
`processing` — ack «взял в работу» (до чтения файла из S3); `done | needs_check | error` — итог.
Условная обязательность — буквально из `allOf` схемы: для `done | needs_check` обязательны `result_s3` и

View File

@ -12,9 +12,9 @@ from typing import Literal, get_args
Pipeline = Literal["tabular", "image"]
NATS_MSG_ID_HEADER = "Nats-Msg-Id"
RESULT_SUBJECT = "recognition.result"
TABULAR_REQUEST_SUBJECT = "recognition.tabular.request"
IMAGE_REQUEST_SUBJECT = "recognition.image.request"
RESULT_SUBJECT = "files.tabular_parsed"
TABULAR_REQUEST_SUBJECT = "files.tabular_uploaded"
IMAGE_REQUEST_SUBJECT = "files.img_uploaded"
REQUEST_SUBJECTS: Mapping[str, str] = {
"tabular": TABULAR_REQUEST_SUBJECT,
@ -23,7 +23,7 @@ REQUEST_SUBJECTS: Mapping[str, str] = {
def request_subject(pipeline: str) -> str:
"""Subject запроса по pipeline: `tabular` → `recognition.tabular.request`, `image` → `recognition.image.request`."""
"""Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`."""
try:
return REQUEST_SUBJECTS[pipeline]
except KeyError:
@ -36,10 +36,13 @@ def request_msg_id(job_id: uuid.UUID | str, attempt: int) -> str:
def ack_msg_id(job_id: uuid.UUID | str, attempt: int) -> str:
"""`Nats-Msg-Id` ack (`status = processing`) на `recognition.result`: `ack:<job_id>:<attempt>`."""
"""`Nats-Msg-Id` ack (`status = processing`) на `files.tabular_parsed`: `ack:<job_id>:<attempt>`."""
return f"ack:{job_id}:{attempt}"
def result_msg_id(job_id: uuid.UUID | str, attempt: int) -> str:
"""`Nats-Msg-Id` результата (`done | needs_check | error`) на `recognition.result`: `result:<job_id>:<attempt>`."""
"""`Nats-Msg-Id` результата (`done | needs_check | error`) на `files.tabular_parsed`.
Форма: `result:<job_id>:<attempt>`.
"""
return f"result:{job_id}:{attempt}"

View File

@ -71,9 +71,9 @@ class Settings(BaseSettings):
server: ServerSettings = ServerSettings()
ROOT_PATH: str = "/"
VERSION: str = "1.0.0"
SERVICE_NAME: str = "scan-gateway" # service.name в конверте recognition.result
SERVICE_NAME: str = "scan-gateway" # service.name в конверте files.tabular_parsed
# NATS: подписка на recognition.image.request и публикация. Выключено — воркер не запускается (dev без NATS)
# NATS: подписка на files.img_uploaded и публикация. Выключено — воркер не запускается (dev без NATS)
NATS_ENABLED: bool = True
PUBLISH_RETRIES: PositiveInt = 3 # повторы публикации результата/ack после первой попытки (всего до 4)
DONE_CACHE_SIZE: PositiveInt = 1000 # размер LRU обработанных job_id (дедупликация повторов)

View File

@ -8,7 +8,7 @@
поэтому `connect()` возвращается только при успехе (вызывающий оборачивает его в задачу или `wait_for`).
После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно
оформлять не нужно.
- `subscribe_requests(cb)` — подписка на `recognition.image.request` в queue group `scan-gateway`: из нескольких
- `subscribe_requests(cb)` — подписка на `files.img_uploaded` в queue group `scan-gateway`: из нескольких
инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py (`msg.data`, `msg.headers`);
исключения `cb` логируются, подписка живёт.
- `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush` (подтверждение сервером) с
@ -209,7 +209,7 @@ class NatsGateway:
# --- подписка ---------------------------------------------------------------------------------------------
async def subscribe_requests(self, cb: MessageHandler, queue: str = REQUEST_QUEUE_GROUP) -> Any:
"""Подписка на `recognition.image.request` в queue group `queue`; возвращает подписку nats-py после
"""Подписка на `files.img_uploaded` в queue group `queue`; возвращает подписку nats-py после
подтверждения сервером (`flush`).
`cb` получает исходное `Msg` (`data`, `headers`); его исключения логируются, подписка не падает.
@ -292,7 +292,7 @@ class NatsGateway:
) from exc
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None:
"""Публикует конверт в `recognition.result`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` —
"""Публикует конверт в `files.tabular_parsed`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` —
`ack:…` для `processing`, `result:…` для итога; `attempt` — из запроса. Ошибки — как у `publish`."""
payload = json.dumps(message.to_wire(), ensure_ascii=False).encode("utf-8")
await self.publish(RESULT_SUBJECT, payload, message_msg_id(message, attempt))

View File

@ -107,7 +107,7 @@ class ObjectStorage(Protocol):
class ResultPublisher(Protocol):
"""Часть `NatsGateway`: публикация конверта в `recognition.result` с `Nats-Msg-Id` и повторами."""
"""Часть `NatsGateway`: публикация конверта в `files.tabular_parsed` с `Nats-Msg-Id` и повторами."""
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ...
@ -181,7 +181,7 @@ class _WaitResult:
class Worker:
"""Обработка запросов `recognition.image.request` (см. докстринг модуля).
"""Обработка запросов `files.img_uploaded` (см. докстринг модуля).
Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для
дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов).

View File

@ -22,12 +22,12 @@
ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора
— по settle (`SETTLE_SECONDS`, 0.2 с).
Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `recognition.result`
Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `files.tabular_parsed`
отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-<uuid>/…`, `raw/<job_id>.json`, `results/…`) и
удаляются.
Подписка шлюза идёт в queue group `scan-gateway`: если в этом NATS живёт другой экземпляр scan-gateway, он может
забрать запрос теста (тест тогда упадёт по таймауту ожидания ack). Перед запуском занятость группы можно проверить
мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `recognition.image.request` быть
мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `files.img_uploaded` быть
не должно. Автоматической защиты в тесте нет намеренно (она была бы гонкой). Недоступная инфраструктура — падение
теста с понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`).
"""
@ -163,7 +163,7 @@ class Envelope:
@dataclass
class Roundtrip:
"""Запущенный шлюз (lifespan приложения), эмулятор и слушатель `recognition.result`."""
"""Запущенный шлюз (lifespan приложения), эмулятор и слушатель `files.tabular_parsed`."""
settings: Settings
app: FastAPI
@ -209,7 +209,7 @@ class Roundtrip:
}
async def publish(self, body: dict[str, Any]) -> None:
"""Публикация как в ядре: `recognition.image.request`, `Nats-Msg-Id = <job_id>:<attempt>`."""
"""Публикация как в ядре: `files.img_uploaded`, `Nats-Msg-Id = <job_id>:<attempt>`."""
payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
headers = {NATS_MSG_ID_HEADER: request_msg_id(body["job_id"], body["attempt"])}
await self.publisher.publish(IMAGE_REQUEST_SUBJECT, payload, headers=headers)
@ -378,7 +378,7 @@ def _settings(monkeypatch: pytest.MonkeyPatch, root: Path, *, cleanup: bool) ->
async def _start_gateway(stack: AsyncExitStack, app: FastAPI) -> None:
"""Lifespan приложения; возврат — когда шлюз подписан на `recognition.image.request`."""
"""Lifespan приложения; возврат — когда шлюз подписан на `files.img_uploaded`."""
await stack.enter_async_context(app.router.lifespan_context(app))
[connect_task] = [t for t in asyncio.all_tasks() if t.get_name() == "nats-connect"]
# задача завершается после SUB с flush: дальше подписка шлюза действует на сервере

View File

@ -629,9 +629,9 @@ def test_ack_and_result_msg_ids() -> None:
def test_request_subject_by_pipeline() -> None:
assert request_subject("tabular") == "recognition.tabular.request"
assert request_subject("image") == "recognition.image.request"
assert RESULT_SUBJECT == "recognition.result"
assert request_subject("tabular") == "files.tabular_uploaded"
assert request_subject("image") == "files.img_uploaded"
assert RESULT_SUBJECT == "files.tabular_parsed"
assert NATS_MSG_ID_HEADER == "Nats-Msg-Id"
@ -643,7 +643,8 @@ def test_request_subject_rejects_unknown_pipeline() -> None:
def test_request_subject_matches_example_pipeline() -> None:
for name in ("request-tabular.json", "request-image.json"):
request = RecognitionRequest.model_validate(_example(name))
assert request_subject(request.pipeline) == f"recognition.{request.pipeline}.request"
expected = {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}[request.pipeline]
assert request_subject(request.pipeline) == expected
def test_integer_valued_float_is_normalized_to_int() -> None:
@ -704,7 +705,7 @@ def test_is_supported_major_agrees_with_schema_pattern(version: str) -> None:
def test_image_request_subject() -> None:
assert IMAGE_REQUEST_SUBJECT == "recognition.image.request"
assert IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
# --- конструкторы конвертов -----------------------------------------------------------------------------------------

View File

@ -253,7 +253,7 @@ async def test_subscribe_requests_uses_image_subject_and_queue_group(gateway: Na
assert REQUEST_QUEUE_GROUP == "scan-gateway"
[subscription] = fake.subscriptions
assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "recognition.image.request"
assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
assert subscription["queue"] == "scan-gateway"
assert fake.flush_calls == 1 # подписка подтверждена сервером до возврата

View File

@ -279,7 +279,7 @@ class Env:
class FakeMsg:
def __init__(self, data: bytes) -> None:
self.data = data
self.subject = "recognition.image.request"
self.subject = "files.img_uploaded"
self.headers: dict[str, str] | None = None

View File

@ -0,0 +1,2 @@
schema: spec-driven
created: 2026-10-08

View File

@ -0,0 +1,14 @@
## Context
Переименование subject'ов по решению владельца 08.10.2026; схема обмена (два subject'а запроса по `pipeline`, один
общий subject ответа) не меняется. Подробности и риски — design сквозного change'а `nats-files-subjects` в docs.
## Decisions
Решения R-1…R-5 приняты в сквозном change'е (docs, `openspec/changes/nats-files-subjects/design.md`) и здесь не
дублируются.
## Decisions made without discussion
Своих решений у этого change'а нет: маршрут (R-1), версия контракта (R-2), DBML (R-3), константы вместо настроек (R-4),
историю не переписываем (R-5) — см. design docs.

View File

@ -0,0 +1,31 @@
## Why
Решение владельца 08.10.2026: subject'ы NATS — «правильно как в старом АРМ». На проде заведены `files.tabular_uploaded`,
`files.img_uploaded`, `files.tabular_parsed`, `files.img_parsed`; используем три, `files.img_parsed` не используем,
старый АРМ гасится. Сквозной change — `nats-files-subjects` в RSHB/arm-ka-docs (issue #74, design с решениями R-1…R-5).
Issue этого репозитория — RSHB/arm-scan-gateway#22.
## What Changes
- `recognition.tabular.request` → `files.tabular_uploaded`, `recognition.image.request` → `files.img_uploaded`,
`recognition.result` → `files.tabular_parsed` (константы `api/src/contract/subjects.py`, докстринги, тесты, README,
`CLAUDE.md`).
- После изменения сервис подписан на `files.img_uploaded` в queue group `scan-gateway`, ack и результат публикует в `files.tabular_parsed` (общий subject ответа, `files.img_parsed` не используется).
- Копии схем `api/contracts/recognition-request.schema.json` и `recognition-result-message.schema.json` — побайтно из
`docs/contracts/` (изменены только описания subject'ов).
## Non-goals
- Тела сообщений, `Nats-Msg-Id`, queue group'ы и `schema_version` не меняются.
- Subject'ы не выносятся в настройки (OQ-83).
## Capabilities
### New Capabilities
### Modified Capabilities
- `image-recognition-gateway`: имена subject'ов запроса и ответа.
## Impact
Выкатывается одновременно с остальными сервисами: сервис со старыми именами не увидит сообщений с новыми.

View File

@ -0,0 +1,63 @@
## MODIFIED Requirements
### Requirement: Подтверждение запроса до обработки
Сервис SHALL быть подписан на `files.img_uploaded` в queue group `scan-gateway`. Получив запрос, проходящий схему,
сервис SHALL до чтения файла из S3 опубликовать в `files.tabular_parsed` конверт `status = processing`
(`service.name = "scan-gateway"`, `Nats-Msg-Id = ack:<job_id>:<attempt>`). Повторный запрос с `job_id`, который уже
обрабатывается или обработан этим процессом, SHALL получить повторный ack, и второй файл во входящий каталог MUST NOT записываться.
#### Scenario: Ack раньше файла
- **WHEN** получен запрос для `aep_5.pdf`
- **THEN** конверт `processing` опубликован раньше, чем файл появился во входящем каталоге
#### Scenario: Повторный запрос
- **WHEN** тот же запрос доставлен повторно во время ожидания ответа ИС
- **THEN** публикуется ещё один ack, во входящем каталоге остаётся один файл `<job_id>.pdf`
### Requirement: Результат по контракту
Сервис SHALL публиковать в `files.tabular_parsed` конверт по `recognition-result-message.schema.json` с
`service.name = "scan-gateway"`, `Nats-Msg-Id = result:<job_id>:<attempt>`; при `status ∈ {done, needs_check}` — файл
результата (`recognition-result.schema.json`, `schema_version = "1.0"` — `CONTRACT_VERSION` сервиса, решение №14; схема допускает любой MINOR) в `results/{file_id}/{job_id}/result.json` и `result_s3`;
при неудаче публикации SHALL повторить её до трёх раз. Нормализатор получает склеенный ответ и SHALL считать файл
успешным, если `error.code == ""` (поле `error` реального успешного ответа — `{"code": "", "message": ""}`, а не `null`;
отсутствующий или `null` `error` ошибкой не считается), и ошибочным, если `error.code` непуст, `parse_error` не `null`
либо корень ответа файла не является объектом JSON (ошибка этого файла — `INVALID_FORMAT`); успех MUST NOT определяться
проверкой «`error` не `null`». Если ошибочны **все** файлы — `status = error`, `error.code` и `error.message` берутся у
первого ошибочного файла по имени (для `parse_error` и не-объекта — `INVALID_FORMAT`); если код ИС этого файла не
входит в перечень кодов контракта, `error.code = UNKNOWN_ERROR`, а исходный код SHALL быть указан в `error.message`.
Если есть хотя бы один успешный файл — документы собираются из успешных файлов по требованию «Нормализация ответа ИС»,
а статус определяется по требованию «Статус результата нормализации». В обоих случаях `metrics.warnings` SHALL
содержать по одной строке `<имя файла>: <код>: <сообщение>` на каждый ошибочный файл, где `<код>` — исходный код ИС как
получен (для `parse_error` и не-объекта — `INVALID_FORMAT`); если предупреждений нет, поле `warnings` SHALL быть опущено.
#### Scenario: Успешный ответ ИС в Срезе 0
- **WHEN** ответ из одного файла с `error = {"code": "", "message": ""}` и непустым `data_ocr` (по образцу `aep_6.json`)
- **THEN** начиная со Среза 3a конверт больше не `NOT_IMPLEMENTED`: поведение — по сценарию «Успешный ответ ИС» ниже
#### Scenario: Успешный ответ ИС
- **WHEN** ответ из одного файла с `error = {"code": "", "message": ""}` и непустым `data_ocr` (`aep_6.json`)
- **THEN** конверт `status = needs_check`, `raw_s3_key = raw/<job_id>.json`, `result_s3` указывает на файл результата с двумя документами (`BALANCE`, `INCOME`)
#### Scenario: Ошибка ИС
- **WHEN** единственный файл ответа `<job_id>.json` содержит `error.code = INVALID_DOCUMENT`
- **THEN** конверт `status = error`, `error.code = INVALID_DOCUMENT`, `error.message` из ответа, `raw_s3_key` заполнен, `metrics.warnings` содержит одну строку `<job_id>.json: INVALID_DOCUMENT: <сообщение>`
#### Scenario: Ошибка одного из документов
- **WHEN** ответ состоит из `<job_id>.json` (успех, `aep_3.json`) и `<job_id>_ofr.json` (`INVALID_DOCUMENT`, `aep_3_ofr.json`)
- **THEN** конверт `status = needs_check`, документы собраны из `<job_id>.json`, `metrics.warnings` содержит строку `<job_id>_ofr.json: INVALID_DOCUMENT: <сообщение>` и предупреждения нормализации
#### Scenario: Все документы ошибочны
- **WHEN** оба файла ответа ошибочны (первый по имени — `INVALID_DOCUMENT`, второй — не JSON)
- **THEN** конверт `status = error`, `error.code = INVALID_DOCUMENT` (код первого ошибочного файла по имени), `metrics.warnings` содержит две строки; файл не JSON в warnings несёт код `INVALID_FORMAT`
#### Scenario: Все файлы не JSON
- **WHEN** единственный файл ответа не разбирается как JSON
- **THEN** конверт `status = error`, `error.code = INVALID_FORMAT`, `metrics.warnings` содержит одну строку `<job_id>.json: INVALID_FORMAT: <сообщение разбора>`
#### Scenario: Корень ответа не объект
- **WHEN** файл ответа — корректный JSON, но корень — массив
- **THEN** этот файл ошибочен с кодом `INVALID_FORMAT`; остальные файлы оцениваются по общим правилам
#### Scenario: Код ошибки ИС вне перечня контракта
- **WHEN** единственный файл ответа содержит `error.code = SOME_NEW_CODE`
- **THEN** конверт `status = error`, `error.code = UNKNOWN_ERROR`, `error.message` содержит `SOME_NEW_CODE`, строка в `metrics.warnings` — `<job_id>.json: SOME_NEW_CODE: <сообщение>`

View File

@ -0,0 +1,11 @@
## 1. Переименование
- [x] 1.1 `api/src/contract/subjects.py`: `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed`
- [x] 1.2 Докстринги, логи, тесты, README, `CLAUDE.md` — новые имена
- [x] 1.3 Копии схем `api/contracts/` побайтно из `docs/contracts/`
- [x] 1.4 Дельта спеки `image-recognition-gateway`
## 2. Проверка
- [x] 2.1 `make check`
- [x] 2.2 Интеграционные тесты NATS