arm-core/openspec/specs/recognition-job-exchange/spec.md

recognition-job-exchange Specification

Purpose

Единственная точка обмена ядра с сервисами распознавания — таблица recognition_job (DECISION-4): outbox публикует запросы в NATS, inbox принимает подтверждения и результаты идемпотентно, потеря сообщений без JetStream компенсируется повторной публикацией и таймаутами, файл результата регистрируется в s3_object.

Requirements

Requirement: Outbox публикует задания из queued

Фоновый публикатор SHALL с периодом OUTBOX_POLL_SECONDS (по умолчанию 5) выбирать задания status = queued (с блокировкой FOR UPDATE SKIP LOCKED), формировать сообщение по recognition-request.schema.json v1.0 из recognition_job ⋈ stored_file ⋈ s3_object и публиковать в subject по pipeline с заголовком Nats-Msg-Id = <job_id>:<attempt>. После успешной публикации задание SHALL получить status = published, published_at, request_msg_id, request_subject; при ошибке публикации задание SHALL остаться queued.

Scenario: Публикация запроса парсеру

• WHEN в recognition_job вставлена строка queued с pipeline = tabular, связанная со stored_file (XLSX) и его s3_object
• THEN в течение 10 с в recognition.tabular.request появляется сообщение, проходящее схему запроса, с job_id, file_id, s3.key исходника, reply_deadline, attempt = 1, а строка переходит в published с заполненным request_msg_id

Scenario: NATS недоступен

• WHEN публикация завершается ошибкой соединения
• THEN задание остаётся queued, ошибка залогирована, следующий цикл повторяет публикацию

Requirement: Ack сервиса переводит задание в processing

Получив в recognition.result конверт со status = processing для задания в published, ядро SHALL установить status = processing, started_at, service_name, service_version и перевести файл задания в stored_file.status = PROCESSING, если current_recognition_job_id файла равен этому заданию. Повторный ack для задания в processing SHALL игнорироваться; ack для задания в другом статусе SHALL логироваться и не менять задание.

Scenario: Сервис подтвердил

• WHEN сервис опубликовал конверт processing с job_id опубликованного задания
• THEN задание имеет status = processing, started_at заполнен, service_name = tabular-parser, файл — PROCESSING

Requirement: Запрос без ack публикуется повторно, затем — dlq

Если задание находится в published дольше ACK_TIMEOUT_SECONDS (по умолчанию 30) без ack, публикатор SHALL опубликовать запрос повторно с тем же Nats-Msg-Id и тем же attempt, обновив published_at и увеличив счётчик metrics.republish_count. После OUTBOX_MAX_REPUBLISH (по умолчанию 3) повторов без ack задание SHALL получить status = dlq, error_code = NO_ACK, finished_at.

Scenario: Сервис недоступен

• WHEN ни один инстанс tabular-parser не подписан, задание опубликовано
• THEN запрос публикуется повторно примерно каждые 30 с; после трёх повторов задание имеет status = dlq, error_code = NO_ACK, request_msg_id не изменился

Scenario: Сервис появился после первого повтора

• WHEN сервис подписался между первой и второй публикацией и прислал ack
• THEN задание переходит в processing, дальнейшие повторы не выполняются

Requirement: Inbox принимает результат идемпотентно

Для конверта со status ∈ {done, needs_check, error} ядро SHALL валидировать его по recognition-result-message.schema.json, найти задание по job_id и: если result_msg_id задания уже равен Nats-Msg-Id сообщения — ничего не менять; иначе зарегистрировать файл результата в s3_object (bucket_role = results, created_by = <service.name>; только при наличии result_s3), обновить задание (status, service_name, service_version, finished_at, metrics, error_code, error_message, result_schema_version, result_s3_object_id, result_msg_id) и зафиксировать транзакцию. При status = error файл задания (если current_recognition_job_id равен заданию) SHALL получить stored_file.status = ERROR; при done/needs_check файл остаётся PROCESSING и SHALL быть обработан циклом маппинга (спека osv-result-mapping); inbox MUST NOT читать файл результата. Неизвестный job_id SHALL быть залогирован. Результат для задания, уже находящегося в финальном статусе (done, needs_check, error, timeout, dlq), SHALL логироваться и не применяться.

Scenario: Успешный результат

• WHEN сервис публикует конверт status = done с result_s3 = {arm-ka, results/<file_id>/<job_id>/result.json} и Nats-Msg-Id = result:<job_id>:1
• THEN задание получает status = done, result_msg_id = result:<job_id>:1, result_s3_object_id указывает на новую строку s3_object с этим ключом, finished_at заполнен; файл остаётся PROCESSING до маппинга

Scenario: Повторная доставка

• WHEN то же сообщение получено второй раз
• THEN строка recognition_job и s3_object не меняются

Scenario: Ошибка сервиса

• WHEN конверт содержит status = error, error = {UNSUPPORTED_FORMAT, …} без result_s3
• THEN задание получает status = error, error_code = UNSUPPORTED_FORMAT, error_message из конверта, result_s3_object_id пуст; файл — ERROR

Scenario: Результат пришёл после таймаута

• WHEN задание уже timeout, и приходит конверт done
• THEN задание не меняется, событие залогировано

Requirement: Таймаут ожидания результата

Фоновый вахтёр SHALL с периодом 30 с переводить задания processing и published с reply_deadline < now() в status = timeout, error_code = TIMEOUT, finished_at = now(), а их файлы (по current_recognition_job_id) — в stored_file.status = ERROR. Публикатор при переводе задания в dlq SHALL так же переводить файл в ERROR. Replay задания (новая попытка, attempt + 1) — вне этого среза.

Scenario: Сервис взял, но не ответил

• WHEN задание в processing и reply_deadline прошёл без результата
• THEN не позднее чем через 30 с задание имеет status = timeout, файл — ERROR с error_code = TIMEOUT

Requirement: Соединение с NATS по шаблону, без JetStream

Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы OPENBAO__NATS__NATS_SEED/NATS_JWT, и без учётных данных, если они пусты; подписываться на recognition.result в queue group arm-core; переподключаться автоматически. Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновые задачи SHALL запускаться при NATS_ENABLED=true.

Scenario: Dev-окружение compose

• WHEN ядро стартует с DEBUG=true, NATS_ENABLED=true и пустыми учётными данными против NATS без JetStream
• THEN подписка активна, /healthcheck отдаёт nats = ok, в логах нет ошибок подключения

Scenario: Два инстанса ядра

• WHEN запущены два инстанса с NATS_ENABLED=true и опубликован один результат
• THEN результат обработан ровно одним инстансом (queue group), строка задания изменена один раз

Scenario: NATS выключен

• WHEN NATS_ENABLED=false
• THEN приложение стартует, фоновые задачи не запускаются, /healthcheck отвечает 200 с nats = off