arm-scan-gateway/openspec/specs/image-recognition-gateway/spec.md

image-recognition-gateway Specification

Purpose

Адаптер внешней ИС Сканирование (Content AI) для PDF/PNG/JPG: принимает и подтверждает задание из NATS без JetStream, передаёт файл через горячие папки, ждёт ответ не дольше таймаута, сохраняет сырой ответ в S3 и отвечает ядру по контракту v1.0. Без БД и фронта.

Requirements

Requirement: Сервис не зависит от БД и фронта

Репозиторий MUST NOT содержать web/, alembic/, src/db/, src/models/, src/repositories/; зависимости SQLAlchemy, asyncpg, alembic, raisa_fastapi_protected_api MUST NOT присутствовать; smbprotocol SHALL остаться. Для старта SHALL хватать OPENBAO__MINIO__*, OPENBAO__NATS__NATS_URL, SCAN_HOTFOLDER_MODE=local, SCAN_HOTFOLDER_PATH.

Scenario: Старт в режиме локальных папок

• WHEN сервис запущен с SCAN_HOTFOLDER_MODE=local и существующим каталогом SCAN_HOTFOLDER_PATH
• THEN GET /healthcheck → 200 с полями uptime, version, nats, hotfolder = ok; OPENBAO__SMB__* не требуются

Requirement: Подтверждение запроса до обработки

Сервис SHALL быть подписан на recognition.image.request в queue group scan-gateway. Получив запрос, проходящий схему, сервис SHALL до чтения файла из S3 опубликовать в recognition.result конверт 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>_aep_5.pdf

Requirement: Передача файла в ИС Сканирование через горячие папки

После подтверждения сервис SHALL прочитать исходник из S3 (s3.bucket/s3.key) и записать его во входящий каталог как <job_id>_<original_name>; временный файл записи SHALL лежать в служебном каталоге <корень>/<SCAN_TMP_DIR>/ (по умолчанию Tmp, на той же шаре/ФС, что Input; создаётся при записи) и переименовываться в Input атомарно, так что во входящем каталоге MUST NOT появляться недописанные и временные файлы. Затем сервис SHALL опрашивать исходящий каталог <SCAN_OUTPUT_DIR>/<job_id>/ каждые SCAN_POLL_SECONDS до min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS). ИС кладёт в этот каталог все распознанные документы задания — по одному *.json на документ. Ответ SHALL считаться готовым, когда в каталоге есть хотя бы один *.json, а набор имён и размеров *.json не менялся SCAN_SETTLE_SECONDS (по умолчанию 2); готовый набор SHALL включать все *.json каталога по возрастанию имени, остальные файлы игнорируются.

Scenario: Файл положен во входящий каталог

• WHEN получен запрос для aep_5.pdf с job_id = 3f1c…
• THEN во входящем каталоге появляется 3f1c…_aep_5.pdf, байт в байт равный объекту S3; во время записи во входящем каталоге нет временных файлов, а Tmp/ существует

Scenario: Ответ из одного файла

• WHEN в Output/3f1c…/ появляется один *.json и набор не меняется SCAN_SETTLE_SECONDS
• THEN опрос прекращается не позднее чем через SCAN_SETTLE_SECONDS + SCAN_POLL_SECONDS после первого опроса, увидевшего окончательный набор (отсчёт стабилизации — от этого опроса, а не от времени изменения файла), ответ из одного файла передаётся дальше по конвейеру

Scenario: Ответ из нескольких файлов

• WHEN в Output/3f1c…/ лежат 3f1c….json и 3f1c…_ofr.json, посторонний файл readme.txt, и набор не меняется SCAN_SETTLE_SECONDS
• THEN дальше по конвейеру передаются оба *.json по возрастанию имени, readme.txt игнорируется

Scenario: Набор ещё пишется

• WHEN размер файла или число файлов в Output/3f1c…/ менялось менее SCAN_SETTLE_SECONDS назад
• THEN ответ не считается готовым, опрос продолжается до дедлайна

Requirement: Сырой ответ сохраняется в S3 до нормализации

Все файлы ответа ИС SHALL быть склеены в один объект {"job_id": "<job_id>", "files": [{"name": "<имя файла>", "response": <разобранный json> | null, "parse_error": "<текст>" | null}, ...]} (файлы по возрастанию имени) и записаны в raw/{job_id}.json того же бакета до вызова нормализатора; файл, не разобранный как JSON, SHALL получить response = null и заполненный parse_error и MUST NOT прерывать задание. Ключ SHALL передаваться в конверте результата полем raw_s3_key (одна строка raw/{job_id}.json); в БД объект не регистрируется (ядро его не создаёт).

Scenario: Нормализатор упал

• WHEN ответ ИС получен, а нормализатор бросил исключение
• THEN объект raw/{job_id}.json существует в S3, конверт содержит raw_s3_key и error.code = UNKNOWN_ERROR

Scenario: Склеенный сырой ответ

• WHEN ИС положила в Output/<job_id>/ файлы <job_id>.json и <job_id>_ofr.json
• THEN raw/{job_id}.json содержит job_id и files из двух элементов (<job_id>.json, затем <job_id>_ofr.json), у каждого response — разобранный ответ, parse_error = null

Scenario: Файл не JSON

• WHEN один из файлов ответа не разбирается как JSON
• THEN в files для него response = null, parse_error непустой; остальные файлы разобраны, задание не падает

Requirement: Результат по контракту

Сервис SHALL публиковать в recognition.result конверт по recognition-result-message.schema.json с service.name = "scan-gateway", Nats-Msg-Id = result:<job_id>:<attempt>; при status ∈ {done, needs_check} — файл результата в results/{file_id}/{job_id}/result.json и result_s3; при неудаче публикации SHALL повторить её до трёх раз. В Срезе 0 нормализатор получает склеенный ответ и 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. Если есть хотя бы один успешный файл — status = error, error.code = NOT_IMPLEMENTED. В обоих случаях metrics.warnings SHALL содержать по одной строке <имя файла>: <код>: <сообщение> на каждый ошибочный файл, где <код> — исходный код ИС как получен (для parse_error и не-объекта — INVALID_FORMAT); если ошибочных файлов нет, поле warnings SHALL быть опущено.

Scenario: Успешный ответ ИС в Срезе 0

• WHEN ответ из одного файла с error = {"code": "", "message": ""} и непустым data_ocr (по образцу aep_6.json)
• THEN конверт status = error, error.code = NOT_IMPLEMENTED, raw_s3_key = raw/<job_id>.json, поле metrics.warnings отсутствует

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 (успех) и <job_id>_ofr.json (INVALID_DOCUMENT) — как у фикстур aep_3.json и aep_3_ofr.json
• THEN конверт status = error, error.code = NOT_IMPLEMENTED, 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: <сообщение>

Requirement: Таймаут ожидания ответа

Если готовый набор ответа (см. «Передача файла…») не появился до дедлайна, сервис SHALL опубликовать конверт status = error, error.code = TIMEOUT; повторная отправка файла в ИС MUST NOT выполняться.

Scenario: ИС молчит

• WHEN запрос с reply_deadline через 10 с, ответ не появляется
• THEN через ≤ 15 с опубликован конверт TIMEOUT, входящий файл остаётся в каталоге

Scenario: Набор не успел настояться до дедлайна

• WHEN в Output/<job_id>/ есть *.json, но набор имён или размеров менялся позже, чем SCAN_SETTLE_SECONDS до дедлайна
• THEN ответ не считается готовым, опубликован конверт TIMEOUT, недостающие файлы не ожидаются

Requirement: Ошибки транспорта превращаются в результат

Сервис SHALL отвечать конвертом status = error с кодом по таблице: запрос не по схеме или MAJOR ≠ 1 → SCHEMA_MISMATCH (ack не публикуется); объект S3 не читается → FILE_READ_ERROR; запись во входящий каталог не удалась → RECOGNITION_FAILED; файл ответа ИС не разбирается как JSON → parse_error этого файла, а если ошибочны все файлы — INVALID_FORMAT; иное исключение → UNKNOWN_ERROR. Во всех случаях конверт SHALL быть опубликован, процесс MUST NOT завершаться.

Scenario: Горячая папка недоступна

• WHEN каталог SCAN_HOTFOLDER_PATH не существует в момент обработки
• THEN конверт RECOGNITION_FAILED с сообщением о недоступности горячей папки; /healthcheck.hotfolder = down

Requirement: Параллельная обработка ограничена

Сервис SHALL обрабатывать не более SCAN_MAX_INFLIGHT (по умолчанию 4) заданий одновременно; задания сверх лимита SHALL ждать слот во внутренней очереди, их ack SHALL быть опубликован сразу при получении.

Scenario: Пять заданий подряд

• WHEN в очереди пять запросов при SCAN_MAX_INFLIGHT = 4
• THEN пять ack'ов опубликованы сразу, четыре файла появляются во входящем каталоге, пятый — после завершения любого из четырёх

Requirement: Копия контракта идентична источнику

api/contracts/*.schema.json SHALL быть идентичны ../docs/contracts/; pydantic-модели SHALL валидировать примеры docs; проверка входит в make check.

Scenario: make check

• WHEN выполняются тесты контракта при доступном ../docs
• THEN идентичность подтверждена, примеры валидны

Requirement: Соединение с NATS без JetStream

При пустых NATS_SEED/NATS_JWT сервис SHALL подключаться без учётных данных, иначе — с nkeys/JWT; MUST NOT требовать stream'ов и consumer'ов JetStream; при разрыве SHALL переподключаться и восстанавливать подписку; /healthcheck.nats SHALL отражать состояние.

Scenario: NATS перезапущен

• WHEN контейнер NATS перезапущен во время работы сервиса
• THEN в течение 20 с после старта NATS подписка восстановлена, nats = ok