tasks nats-env-config: check-full зелёный (3.2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -67,7 +67,7 @@
   (`docker build`; убран из `.gitignore` шаблона). CI в репозитории нет: перед PR — `make check-full`. Без Nexus:
   `make venv` ставит зависимости без `raisa_*`, npm-пакеты `raisa-*` — `optionalDependencies`.
 - **NATS без JetStream** (`NATS_ENABLED=true`): `core/nats.py` — подключение в фоне (supervisor повторяет попытки),
-  подписка queue group `arm-core` на `files.tabular_parsed`. Обмен — `services/recognition_exchange.py`: `publish_pending`
+  подписка queue group `arm-core` на `files.tabular_parsed` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT). Обмен — `services/recognition_exchange.py`: `publish_pending`
   (outbox), `expire_deadlines` (вахтёр сроков), `handle_message` (inbox: ack — конверт `processing`, результат);
   публикатор и вахтёр — фоновые задачи lifespan. Запрос без ack повторно публикуется (`ACK_TIMEOUT_SECONDS`=30 ×
   `OUTBOX_MAX_REPUBLISH`=3, счётчик — `recognition_job.metrics.republish_count`) → `dlq`; дедуп — `request_msg_id` /

README.md

@@ -129,7 +129,7 @@ curl -s localhost:8000/healthcheck
 ```
 
 ```json
-{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok"}
+{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok","subjects":{"tabular_request":"files.tabular_uploaded","image_request":"files.img_uploaded","result":"files.tabular_parsed"}}
 ```
 
 `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, `off` — выключен.
@@ -922,6 +922,9 @@ email без учёта регистра проверяет только сер
 | `OPENBAO__NATS__NATS_URL` | — (обязательно) | адрес NATS (`nats://localhost:4222`) |
 | `OPENBAO__NATS__NATS_SEED` | пусто | nkey seed; пусто — подключение без учётных данных (compose, тесты). Задаётся вместе с `NATS_JWT` или не задаётся вовсе |
 | `OPENBAO__NATS__NATS_JWT` | пусто | JWT NATS; см. `NATS_SEED` |
+| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса распознавания таблиц (публикация) |
+| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса распознавания изображений (публикация) |
+| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject ответов (подписка, queue group `arm-core`). Три имени различны, без пробелов, `*`, `>` и пустых токенов; нарушение — отказ старта. Действующие значения — в логе старта и в `/healthcheck.subjects` |
 | `OPENBAO__SMB__SMB_HOST` | пусто | SMB из шаблона, необязателен (ядру в Срезе 0 не нужен). Пусто — сессия SMB не регистрируется; задан — регистрация при старте (в отдельном потоке) и сброс кэша соединений при остановке |
 | `OPENBAO__SMB__SMB_PATH` | пусто | путь SMB (в Срезе 0 не используется) |
 | `OPENBAO__SMB__SMB_LOGIN` | пусто | логин SMB |
@@ -1001,7 +1004,7 @@ tabular-parser (MinIO) → inbox → маппинг → `GET /files/{id}` = `PRO
 
 ```bash
 docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services up -d --build --no-deps tabular-parser
-curl -s localhost:8001/healthcheck      # {"uptime": "...", "version": "...", "nats": "ok"}
+curl -s localhost:8001/healthcheck      # {"uptime": "...", "version": "...", "nats": "ok", "subjects": {"tabular_request": "...", "result": "..."}}
 make e2e                                # PASSED
 docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services stop tabular-parser
 ```

api/.env.example

@@ -37,6 +37,10 @@ OPENBAO__MINIO__MINIO_SECURE=false
 OPENBAO__NATS__NATS_URL=nats://localhost:4222
 OPENBAO__NATS__NATS_SEED=
 OPENBAO__NATS__NATS_JWT=
+# subject'ы обмена — по окружению (прод/тест), у трёх сервисов одинаковые; умолчания — имена контракта
+OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=files.tabular_uploaded
+OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=files.img_uploaded
+OPENBAO__NATS__NATS_RESULT_SUBJECT=files.tabular_parsed
 
 # SMB необязателен (ядру в Срезе 0 не нужен): пустой SMB_HOST — сессия SMB не регистрируется
 OPENBAO__SMB__SMB_HOST=

api/.env.test

@@ -23,6 +23,9 @@ OPENBAO__MINIO__MINIO_SECURE=false
 OPENBAO__NATS__NATS_URL=nats://localhost:4222
 OPENBAO__NATS__NATS_SEED=
 OPENBAO__NATS__NATS_JWT=
+OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=files.tabular_uploaded
+OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=files.img_uploaded
+OPENBAO__NATS__NATS_RESULT_SUBJECT=files.tabular_parsed
 
 # SMB необязателен: пустой SMB_HOST — сессия SMB не регистрируется
 OPENBAO__SMB__SMB_HOST=

api/contracts/recognition-request.schema.json

@@ -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 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.",
+  "description": "Тело сообщения на subject запроса; subject — настройка окружения и должен соответствовать полю pipeline (tabular → NATS_TABULAR_REQUEST_SUBJECT, по умолчанию files.tabular_uploaded; image → NATS_IMAGE_REQUEST_SUBJECT, по умолчанию 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"],

api/contracts/recognition-result-message.schema.json

@@ -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 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).",
+  "description": "Тело сообщения на subject ответа (настройка окружения NATS_RESULT_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"],

api/src/app.py

@@ -1,17 +1,19 @@
 """Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода.
 
 Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core
-NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `files.tabular_parsed`
+NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка на subject ответов (по
-в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается),
+умолчанию `files.tabular_parsed`) в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при
-middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет РАИСА лишь условно импортируется,
+пустом `SMB_HOST` не подключается), middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет
-middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API
+РАИСА лишь условно импортируется, middleware не подключается), healthcheck (uptime, version, nats: ok|down|off,
-(`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`.
+subjects), раздача собранного SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66).
+Приложение собирает `create_app(settings)`.
 Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
 состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор,
 вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов
 (`src/services/result_mapping.py`, Срез 2c D-1) — и останавливает их до закрытия NATS. Сообщения
-`files.tabular_parsed` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий
+subject ответов (по умолчанию `files.tabular_parsed`; ack и результаты сервисов) обрабатывает inbox
-приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1).
+`handle_message` того же модуля с фабрикой сессий приложения (inbox меняет задание и статус файла; маппинг
+результата в доменные таблицы — отдельный цикл, Срез 2c D-1).
 Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного
 логируется и не отменяет остальные. Логгер настраивает `setup_logger` при импорте модуля: без значений локальных
 переменных в трейсбеках (#63).
@@ -43,7 +45,7 @@ from src.core.clients import S3Storage
 from src.core.config import Settings, get_settings
 from src.core.exception_handlers import register_exception_handlers
 from src.core.logger import setup_logger
-from src.core.nats import MessageHandler, NatsConnection
+from src.core.nats import MessageHandler, NatsConnection, used_subjects
 from src.db.db import async_session_maker, engine, session_transaction
 from src.services.recognition_exchange import ExchangeTasks, handle_message
 from src.services.session_audit import SessionAudit
@@ -109,13 +111,14 @@ async def _connect_s3() -> S3Storage | None:
 
 
 def _result_handler() -> MessageHandler:
-    """Inbox `files.tabular_parsed`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл,
+    """Inbox subject ответов (по умолчанию `files.tabular_parsed`): `handle_message` с фабрикой сессий приложения
-    Срез 2c design D-1)."""
+    (маппинг результата — отдельный цикл, Срез 2c design D-1)."""
     return partial(handle_message, session_factory=async_session_maker)
 
 
 async def _subscribe_results(nats: NatsConnection) -> None:
-    """Подписки после первого подключения к NATS: `files.tabular_parsed` в queue group `arm-core`."""
+    """Подписки после первого подключения к NATS: subject ответов (по умолчанию `files.tabular_parsed`) в queue
+    group `arm-core`."""
     await nats.subscribe_results(_result_handler())
 
 
@@ -183,6 +186,7 @@ async def lifespan(app: FastAPI):
     app.state.nats = None
     app.state.exchange = None
 
+    logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS))
     await _register_smb()
 
     try:
@@ -241,7 +245,12 @@ def create_app(app_settings: Settings, *, web_dir: str = "web") -> FastAPI:
         if app_settings.NATS_ENABLED:
             nc = getattr(request.app.state, "nats", None)
             nats_status = "ok" if nc is not None and nc.is_connected else "down"
-        return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status}
+        return {
+            "uptime": UP_TIME,
+            "version": VERSION,
+            "nats": nats_status,
+            "subjects": used_subjects(app_settings.OPENBAO.NATS),
+        }
 
     application.include_router(build_api_router(app_settings))
     # аудит входа (OQ-64): LOGIN первого запроса и LOGIN_FAILED — в отдельной транзакции, множество входов — на процесс

api/src/contract/subjects.py

@@ -22,10 +22,12 @@ REQUEST_SUBJECTS: Mapping[str, str] = {
 }
 
 
-def request_subject(pipeline: str) -> str:
+def request_subject(pipeline: str, subjects: Mapping[str, str] = REQUEST_SUBJECTS) -> str:
-    """Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`."""
+    """Subject запроса по pipeline. `subjects` — карта `pipeline → subject` окружения
+    (`NATSSettings.request_subjects()`); по умолчанию — имена контракта: `tabular` → `files.tabular_uploaded`,
+    `image` → `files.img_uploaded`."""
     try:
-        return REQUEST_SUBJECTS[pipeline]
+        return subjects[pipeline]
     except KeyError:
         raise ValueError(f"неизвестный pipeline {pipeline!r}: допустимы {', '.join(get_args(Pipeline))}") from None
 

api/src/core/config.py

@@ -3,19 +3,27 @@ import os
 import re
 from typing import Annotated, Any
 
-from pydantic import NonNegativeInt, PositiveInt, field_validator, model_validator
+from pydantic import NonNegativeInt, PositiveInt, SecretStr, ValidationInfo, field_validator, model_validator
 from pydantic_settings import BaseSettings, NoDecode, PydanticBaseSettingsSource, SettingsConfigDict
 
+from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT
 from src.core.settings_sources import OpenBaoAppSettingsSource
 
 DB_SCHEMA_RE = re.compile(r"[a-z_][a-z0-9_]{0,62}")
+# Subject NATS без шаблонов: непустые токены через точку, без пробельных символов, `*` и `>` (docs nats-env-config)
+NATS_SUBJECT_RE = re.compile(r"[^\s.*>]+(?:\.[^\s.*>]+)*")
+NATS_SUBJECT_FIELDS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
 
 
 class DBSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     DATABASE_URL: str
 
 
 class MinioSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     MINIO_URL: str
     MINIO_ADMIN_LOGIN: str
     MINIO_ADMIN_PASSWORD: str
@@ -24,15 +32,57 @@ class MinioSettings(BaseSettings):
 
 
 class NATSSettings(BaseSettings):
+    """Подключение и subject'ы NATS — секреты окружения `OPENBAO__NATS__*` (прод и тест — свои наборы).
+
+    Пустые `NATS_SEED` и `NATS_JWT` — подключение без nkeys (локальная разработка, тесты). Subject'ы по умолчанию —
+    имена контракта v1.0; у ядра и сервисов одного окружения они обязаны совпадать (docs, change nats-env-config).
+    """
+
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     NATS_URL: str
-    NATS_SEED: str = ""  # пусто — подключение к NATS без nkeys (локальная разработка, тесты)
+    NATS_SEED: SecretStr = SecretStr("")  # пусто — подключение к NATS без nkeys (локальная разработка, тесты)
-    NATS_JWT: str = ""
+    NATS_JWT: SecretStr = SecretStr("")
+    NATS_TABULAR_REQUEST_SUBJECT: str = TABULAR_REQUEST_SUBJECT
+    NATS_IMAGE_REQUEST_SUBJECT: str = IMAGE_REQUEST_SUBJECT
+    NATS_RESULT_SUBJECT: str = RESULT_SUBJECT
+
+    @field_validator(*NATS_SUBJECT_FIELDS)
+    @classmethod
+    def _check_subject(cls, value: str, info: ValidationInfo) -> str:
+        """Конкретный subject без шаблонов: иначе подписка молча слушала бы чужой трафик или ничего."""
+        if not NATS_SUBJECT_RE.fullmatch(value):
+            raise ValueError(
+                f"{info.field_name}={value!r}: нужен subject NATS из непустых токенов через точку, "
+                "без пробелов, '*' и '>'"
+            )
+        return value
+
+    @model_validator(mode="after")
+    def _check_distinct_subjects(self) -> "NATSSettings":
+        """Три subject'а попарно различны: совпадение запроса и ответа — сервис получает свои ответы, двух запросов —
+        каждый запрос получают оба сервиса."""
+        seen: dict[str, str] = {}
+        for field in NATS_SUBJECT_FIELDS:
+            subject = getattr(self, field)
+            if subject in seen:
+                raise ValueError(
+                    f"{seen[subject]} и {field} совпадают ({subject!r}): subject'ы NATS должны различаться"
+                )
+            seen[subject] = field
+        return self
+
+    def request_subjects(self) -> dict[str, str]:
+        """Subject запроса по `pipeline` (`tabular`, `image`) — аргумент `request_subject` контракта."""
+        return {"tabular": self.NATS_TABULAR_REQUEST_SUBJECT, "image": self.NATS_IMAGE_REQUEST_SUBJECT}
 
 
 class SMBSettings(BaseSettings):
     """SMB (горячие папки ИС Сканирование, шаблон). Необязательны: ядру в Срезе 0 SMB не нужен; пустой `SMB_HOST` —
     сессия SMB не регистрируется."""
 
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     SMB_HOST: str = ""
     SMB_PATH: str = ""
     SMB_LOGIN: str = ""
@@ -40,6 +90,8 @@ class SMBSettings(BaseSettings):
 
 
 class OpenBAOSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     DATABASE: DBSettings
     MINIO: MinioSettings
     NATS: NATSSettings
@@ -57,6 +109,8 @@ class OpenBAOSettings(BaseSettings):
 
 
 class ServerSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     HOST: str = "0.0.0.0"
     PORT: int = 8000
 
@@ -162,6 +216,7 @@ class Settings(BaseSettings):
         env_file=os.environ.get("ENV_FILE", ".env"),
         env_nested_delimiter="__",
         nested_model_default_partial_update=False,
+        hide_input_in_errors=True,
     )
 
 

api/src/core/nats.py

@@ -4,8 +4,9 @@
 - Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных.
 - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется
   сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`).
-- `subscribe_results(handler)` — подписка на `files.tabular_parsed` в queue group `arm-core`: из нескольких инстансов
+- `subscribe_results(handler)` — подписка на `NATS_RESULT_SUBJECT` настроек (умолчание `files.tabular_parsed`) в queue
-  ядра сообщение получает ровно один. Исключения обработчика логируются, подписка не падает.
+  group `arm-core`: из нескольких инстансов ядра сообщение получает ровно один. Исключения обработчика
+  логируются, подписка не падает.
 - Stream'ов и consumer'ов JetStream нет. Переподключение после разрыва — встроенное в nats-py
   (`max_reconnect_attempts=-1`, `reconnect_time_wait=2`), подписки nats-py восстанавливает сам.
 - Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с
@@ -27,7 +28,7 @@ from loguru import logger
 from nats.aio.client import Client as NATS
 from nats.aio.msg import Msg
 
-from src.contract.subjects import NATS_MSG_ID_HEADER, RESULT_SUBJECT
+from src.contract.subjects import NATS_MSG_ID_HEADER
 from src.core.config import NATSSettings
 
 RESULT_QUEUE_GROUP = "arm-core"
@@ -60,14 +61,23 @@ class NatsPublishError(NatsError):
     """Публикация не подтверждена: соединения нет или сервер не ответил на flush."""
 
 
+def used_subjects(settings: NATSSettings) -> dict[str, str]:
+    """Subject'ы ядра для лога старта и `/healthcheck.subjects`: публикует оба запроса, подписано на ответ."""
+    return {
+        "tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT,
+        "image_request": settings.NATS_IMAGE_REQUEST_SUBJECT,
+        "result": settings.NATS_RESULT_SUBJECT,
+    }
+
+
 def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
     """Параметры аутентификации для `Client.connect`.
 
     Пустые `NATS_SEED` и `NATS_JWT` — подключение без учётных данных (docker compose, тесты): `{}`. Заданы оба —
     колбэки nkeys/JWT, как в шаблоне. Задан только один или seed не читается — `NatsConfigError`.
     """
-    seed = settings.NATS_SEED
+    seed = settings.NATS_SEED.get_secret_value()
-    jwt = settings.NATS_JWT
+    jwt = settings.NATS_JWT.get_secret_value()
     if not seed and not jwt:
         return {}
     if not seed or not jwt:
@@ -95,7 +105,7 @@ class NatsConnection:
 
     `on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении
     повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов —
-    по умолчанию `files.tabular_parsed` и `arm-core`.
+    по умолчанию `files.tabular_parsed` (фактический subject задаётся настройками окружения) и `arm-core`.
     """
 
     def __init__(  # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги
@@ -103,7 +113,7 @@ class NatsConnection:
         settings: NATSSettings,
         *,
         on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None,
-        result_subject: str = RESULT_SUBJECT,
+        result_subject: str | None = None,
         queue: str = RESULT_QUEUE_GROUP,
         client_factory: Callable[[], Any] = NATS,
         sleep: SleepFn = asyncio.sleep,
@@ -114,7 +124,7 @@ class NatsConnection:
     ) -> None:
         self._settings = settings
         self._on_connected = on_connected
-        self.result_subject = result_subject
+        self.result_subject = result_subject or settings.NATS_RESULT_SUBJECT
         self.queue = queue
         self._client_factory = client_factory
         self._sleep = sleep

api/src/services/primary_files.py

@@ -480,7 +480,7 @@ class PrimaryFilesService:
             id=job_id,
             file_id=file_id,
             pipeline=detected.pipeline,
-            request_subject=request_subject(detected.pipeline.value),
+            request_subject=request_subject(detected.pipeline.value, self._settings.OPENBAO.NATS.request_subjects()),
             schema_version=CONTRACT_VERSION,
             status=RecognitionStatus.queued,
             attempt=1,

api/src/services/recognition_exchange.py

@@ -26,10 +26,10 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро
 
 Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)).
 
-Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `files.tabular_parsed`
+Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки на subject ответов (по
-(`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO:
+умолчанию `files.tabular_parsed`; `src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая
-`SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit,
+транзакция без сетевого IO: `SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор
-применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`.
+держит строку до своего commit, применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`.
 - ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING`
   (R-5: только из `UPLOADED`); повторный ack для `processing` — тихо игнорируется; ack для другого статуса — warning,
   задание и файл не меняются.
@@ -64,6 +64,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from src.contract import (
     CONTRACT_VERSION,
     NATS_MSG_ID_HEADER,
+    REQUEST_SUBJECTS,
     MessageStatus,
     Metrics,
     RecognitionRequest,
@@ -211,6 +212,7 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
     ack_cutoff: datetime,
     pass_started: datetime,
     max_republish: int,
+    request_subjects: Mapping[str, str],
 ) -> bool:
     """Обработать одно задание в транзакции `session`; `False` — подходящих заданий больше нет."""
     row = (await session.execute(_candidate_stmt(pass_started, ack_cutoff))).one_or_none()
@@ -244,7 +246,7 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
     reply_deadline = row.reply_deadline or now + DEFAULT_REPLY_TIMEOUT[row.pipeline]
     try:
         request = _build_request(row, reply_deadline)
-        subject = request_subject(row.pipeline.value)
+        subject = request_subject(row.pipeline.value, request_subjects)
     except (ValidationError, ValueError) as exc:
         await session.execute(
             _job_update(row.id, row.status).values(
@@ -297,7 +299,7 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
         metrics = base.op("||", return_type=JSONB)(literal({REPUBLISH_COUNT_KEY: count}, JSONB))
         await session.execute(
             _job_update(row.id, RecognitionStatus.published).values(
-                published_at=now, reply_deadline=reply_deadline, metrics=metrics
+                published_at=now, reply_deadline=reply_deadline, metrics=metrics, request_subject=subject
             )
         )
         report.republished += 1
@@ -320,6 +322,7 @@ async def publish_pending(  # noqa: PLR0913 - параметры прохода
     max_republish: int,
     batch_size: int = PUBLISH_BATCH_SIZE,
     now: Clock | None = None,
+    request_subjects: Mapping[str, str] = REQUEST_SUBJECTS,
 ) -> PublishReport:
     """Один проход outbox-публикатора (spec «Outbox публикует задания из `queued`», «Запрос без ack публикуется
     повторно, затем — `dlq`»; design D-6).
@@ -346,6 +349,7 @@ async def publish_pending(  # noqa: PLR0913 - параметры прохода
                     ack_cutoff=ack_cutoff,
                     pass_started=pass_started,
                     max_republish=max_republish,
+                    request_subjects=request_subjects,
                 )
         except NatsPublishError as exc:
             report.failed += 1
@@ -402,11 +406,18 @@ async def run_publisher_loop(  # noqa: PLR0913 - настройки цикла
     ack_timeout: timedelta,
     max_republish: int,
     sleep: SleepFn = asyncio.sleep,
+    request_subjects: Mapping[str, str] = REQUEST_SUBJECTS,
 ) -> None:
     """Бесконечный цикл публикатора с периодом `poll_seconds`; ошибка прохода логируется, цикл продолжается."""
     while True:
         try:
-            await publish_pending(session_factory, nats, ack_timeout=ack_timeout, max_republish=max_republish)
+            await publish_pending(
+                session_factory,
+                nats,
+                ack_timeout=ack_timeout,
+                max_republish=max_republish,
+                request_subjects=request_subjects,
+            )
         except Exception as exc:
             logger.exception(
                 "Ошибка прохода outbox-публикатора. error_category=outbox_error, error_type={error_type}, "
@@ -459,6 +470,7 @@ class ExchangeTasks:
                 poll_seconds=settings.OUTBOX_POLL_SECONDS,
                 ack_timeout=timedelta(seconds=settings.ACK_TIMEOUT_SECONDS),
                 max_republish=settings.OUTBOX_MAX_REPUBLISH,
+                request_subjects=settings.OPENBAO.NATS.request_subjects(),
             ),
             name="recognition-outbox-publisher",
         )
@@ -496,7 +508,7 @@ class ExchangeTasks:
                 )
 
 
-# --- inbox: files.tabular_parsed ----------------------------------------------------------------------------------
+# --- inbox: subject ответов ----------------------------------------------------------------------------------
 
 # Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется
 FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset(
@@ -740,7 +752,7 @@ async def _apply_message(
     job = await _lock_job(session, message.job_id)
     if job is None:
         logger.warning(
-            "Сообщение files.tabular_parsed для неизвестного задания, не применено. error_category=unknown_job, "
+            "Сообщение на subject ответа для неизвестного задания, не применено. error_category=unknown_job, "
             "job_id={job_id}, status={status}",
             job_id=message.job_id,
             status=message.status.value,
@@ -748,7 +760,7 @@ async def _apply_message(
         return
     if job.file_id != message.file_id:
         logger.warning(
-            "Сообщение files.tabular_parsed: file_id не совпадает с заданием, не применено. "
+            "Сообщение на subject ответа: file_id не совпадает с заданием, не применено. "
             "error_category=file_mismatch, job_id={job_id}, file_id={file_id}, message_file_id={message_file_id}",
             job_id=job.id,
             file_id=job.file_id,
@@ -760,7 +772,7 @@ async def _apply_message(
     if msg_id is None:
         msg_id = ack_msg_id(job.id, job.attempt) if is_ack else result_msg_id(job.id, job.attempt)
         logger.warning(
-            "Сообщение files.tabular_parsed без заголовка Nats-Msg-Id: использован производный id. "
+            "Сообщение на subject ответа без заголовка Nats-Msg-Id: использован производный id. "
             "error_category=missing_msg_id, job_id={job_id}, msg_id={msg_id}",
             job_id=job.id,
             msg_id=msg_id,
@@ -778,8 +790,9 @@ async def handle_message(
     session_factory: async_sessionmaker[AsyncSession],
     now: Clock | None = None,
 ) -> None:
-    """Обработчик `files.tabular_parsed` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает
+    """Обработчик subject ответов (по умолчанию `files.tabular_parsed`; spec «Ack сервиса переводит задание в
-    результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля.
+    `processing`», «Inbox принимает результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring
+    модуля.
 
     Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл),
     `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7,

api/tests/test_api_files_upload.py

@@ -22,7 +22,7 @@ from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.api.dependencies.directories import get_app_settings
-from src.core.config import get_settings
+from src.core.config import NATSSettings, get_settings
 from src.models.access import AppUser
 from src.models.audit_ex import AuditEvent
 from src.models.consolidation import Consolidation, ConsolidationStatusHistory, ConsolidationVersion
@@ -210,6 +210,32 @@ async def test_real_osv_is_accepted_and_published_from_outbox(
 # --- поля записей, имя файла, аудит ------------------------------------------------------------------------------
 
 
+@pytest.mark.integration
+async def test_job_request_subject_comes_from_environment_settings(
+    app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
+) -> None:
+    """Review Focus 4: `recognition_job.request_subject` берётся из настроек окружения, а не из констант контракта."""
+    base = get_settings()
+    nats = NATSSettings(
+        **{
+            **base.OPENBAO.NATS.model_dump(),
+            "NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded",
+        }
+    )
+    app.dependency_overrides[get_app_settings] = lambda: base.model_copy(
+        update={"OPENBAO": base.OPENBAO.model_copy(update={"NATS": nats})}
+    )
+    entity = await _rostov_entity(seeded)
+    client = await auth_client(KA_RF)
+
+    response = await _post(client, entity.id, ("osv.xls", _xls("subject")))
+
+    assert response.status_code == HTTP_CREATED, response.text
+    job = await seeded.get(RecognitionJob, uuid.UUID(response.json()["accepted"][0]["job_id"]))
+    assert job is not None
+    assert job.request_subject == "test.files.tabular_uploaded"
+
+
 @pytest.mark.integration
 async def test_hostile_name_gives_flat_key_and_full_records(
     auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage

api/tests/test_app.py

@@ -18,6 +18,7 @@ import pytest
 from fastapi import Request
 from fastapi.routing import APIRoute
 from loguru import logger
+from pydantic import SecretStr
 
 from src.api.router import router
 from src.core.config import Settings
@@ -47,6 +48,49 @@ async def test_healthcheck_nats_off_when_disabled(client: httpx.AsyncClient, mon
     assert {"uptime", "version"} <= set(body)
 
 
+def _expected_subjects() -> dict[str, str]:
+    nats = app_module.settings.OPENBAO.NATS
+    return {
+        "tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT,
+        "image_request": nats.NATS_IMAGE_REQUEST_SUBJECT,
+        "result": nats.NATS_RESULT_SUBJECT,
+    }
+
+
+@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
+async def test_healthcheck_subjects(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool):
+    """Review Focus 5: subjects отдаются при любом состоянии NATS, JWT в ответе нет."""
+    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
+    body = (await client.get("/healthcheck")).json()
+    assert body["subjects"] == _expected_subjects()
+    assert "NATS_JWT" not in str(body)
+    assert "jwt" not in str(body).lower()
+
+
+@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
+async def test_healthcheck_returns_non_default_subjects_without_jwt(
+    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool
+):
+    """T3.1: отдаются именно настроенные (не умолчательные) subject'ы; непустой JWT в тело не попадает."""
+    jwt = "JWTVALUEMARK"
+    nats = app_module.settings.OPENBAO.NATS
+    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
+    monkeypatch.setattr(nats, "NATS_TABULAR_REQUEST_SUBJECT", "env.files.tabular_uploaded")
+    monkeypatch.setattr(nats, "NATS_IMAGE_REQUEST_SUBJECT", "env.files.img_uploaded")
+    monkeypatch.setattr(nats, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
+    monkeypatch.setattr(nats, "NATS_JWT", SecretStr(jwt))
+    assert nats.NATS_JWT.get_secret_value() == jwt
+
+    response = await client.get("/healthcheck")
+
+    assert response.json()["subjects"] == {
+        "tabular_request": "env.files.tabular_uploaded",
+        "image_request": "env.files.img_uploaded",
+        "result": "env.files.tabular_parsed",
+    }
+    assert jwt not in response.text
+
+
 async def test_healthcheck_nats_down_when_enabled_without_connection(
     client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
 ):
@@ -220,6 +264,26 @@ async def test_lifespan_does_not_wait_for_nats_and_healthcheck_is_down(monkeypat
     assert not nats.is_connected
 
 
+@pytest.mark.usefixtures("_no_smb")
+async def test_lifespan_logs_nats_subjects_at_start(monkeypatch: pytest.MonkeyPatch):
+    """Tasks 1.2: при старте в лог пишутся используемые subject'ы NATS."""
+    records: list[Any] = []
+    handler_id = logger.add(lambda message: records.append(message.record), level="INFO")
+    monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3)
+    monkeypatch.setattr(app_module, "_create_nats", lambda: None)
+    monkeypatch.setattr(app_module.settings.OPENBAO.NATS, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
+    try:
+        async with app_module.lifespan(app_module.app):
+            pass
+    finally:
+        logger.remove(handler_id)
+
+    messages = [r["message"] for r in records if "Subject'ы NATS" in r["message"]]
+    assert len(messages) == 1
+    assert "env.files.tabular_parsed" in messages[0]
+    assert "files.tabular_uploaded" in messages[0]
+
+
 def test_create_nats_disabled_returns_none(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
     assert app_module._create_nats() is None

api/tests/test_config.py

@@ -36,15 +36,7 @@ def test_env_test_is_read(clean_env: pytest.MonkeyPatch) -> None:
 
 def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
     """Значения по умолчанию классов настроек: env-файла нет, заданы только обязательные OPENBAO__* (SMB не нужен)."""
-    required = {
+    for key, value in REQUIRED_ENV.items():
-        "OPENBAO__DATABASE__DATABASE_URL": "postgresql+asyncpg://u:p@localhost:5432/d",
-        "OPENBAO__MINIO__MINIO_URL": "localhost:9000",
-        "OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "login",
-        "OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "password",
-        "OPENBAO__MINIO__MINIO_BUCKET": "bucket",
-        "OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
-    }
-    for key, value in required.items():
         clean_env.setenv(key, value)
     s = Settings(_env_file=None)  # type: ignore[call-arg]
     assert s.DEBUG is False
@@ -58,8 +50,8 @@ def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
     assert s.NATS_ENABLED is False
     assert (s.OUTBOX_POLL_SECONDS, s.ACK_TIMEOUT_SECONDS, s.OUTBOX_MAX_REPUBLISH) == (5, 30, 3)
     assert (s.UPLOAD_MAX_BYTES, s.MAPPING_POLL_SECONDS, s.MAPPING_MAX_ATTEMPTS) == (50 * 1024 * 1024, 5, 3)
-    assert s.OPENBAO.NATS.NATS_SEED == ""
+    assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
-    assert s.OPENBAO.NATS.NATS_JWT == ""
+    assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
     assert s.OPENBAO.MINIO.MINIO_SECURE is True
     assert (s.server.HOST, s.server.PORT) == ("0.0.0.0", 8000)  # noqa: S104 - по умолчанию, как в контейнере
     smb = s.OPENBAO.SMB
@@ -69,8 +61,8 @@ def test_defaults(clean_env: pytest.MonkeyPatch) -> None:
 def test_empty_nats_seed_and_jwt_allowed(clean_env: pytest.MonkeyPatch) -> None:
     """Пустые NATS_SEED/NATS_JWT допустимы (подключение без nkeys)."""
     s = Settings(_env_file=API_DIR / ".env.example")  # type: ignore[call-arg]
-    assert s.OPENBAO.NATS.NATS_SEED == ""
+    assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
-    assert s.OPENBAO.NATS.NATS_JWT == ""
+    assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
     assert s.NATS_ENABLED is True
     assert s.OPENBAO.SMB.SMB_HOST == ""
     assert (s.server.HOST, s.server.PORT) == ("127.0.0.1", 8000)  # dev-запуск — только loopback (M-11)
@@ -155,3 +147,110 @@ def test_cors_origins_from_env(clean_env: pytest.MonkeyPatch, raw: str, expected
     clean_env.setenv("CORS_ORIGINS", raw)
     s = Settings(_env_file=API_DIR / ".env.test")  # type: ignore[call-arg]
     assert s.CORS_ORIGINS == expected
+
+
+@pytest.fixture
+def settings_from_env(clean_env: pytest.MonkeyPatch):
+    def make(**env: str | None) -> Settings:
+        for key, value in {**REQUIRED_ENV, **env}.items():
+            if value is None:
+                clean_env.delenv(key, raising=False)
+            else:
+                clean_env.setenv(key, value)
+        return Settings(_env_file=None)  # type: ignore[call-arg]
+
+    return make
+
+
+JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
+SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
+
+
+def test_nats_subjects_default_to_contract_names(settings_from_env) -> None:
+    nats = settings_from_env().OPENBAO.NATS
+    assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
+    assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
+    assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
+    assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
+
+
+def test_nats_subjects_from_openbao(settings_from_env) -> None:
+    nats = settings_from_env(
+        OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
+        OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
+        OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
+    ).OPENBAO.NATS
+    assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
+    assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
+
+
+SUBJECT_KEYS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
+
+
+@pytest.mark.parametrize("key", SUBJECT_KEYS)
+@pytest.mark.parametrize(
+    "value",
+    ["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
+)
+def test_invalid_subject_rejected(settings_from_env, key: str, value: str) -> None:
+    with pytest.raises(ValidationError, match=key):
+        settings_from_env(**{f"OPENBAO__NATS__{key}": value})
+
+
+@pytest.mark.parametrize(
+    ("first", "second"),
+    [
+        ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT"),
+        ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
+        ("NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"),
+    ],
+)
+def test_equal_subjects_rejected(settings_from_env, first: str, second: str) -> None:
+    with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
+        settings_from_env(**{f"OPENBAO__NATS__{first}": "files.same", f"OPENBAO__NATS__{second}": "files.same"})
+
+
+def test_invalid_subject_error_hides_nats_secrets(settings_from_env) -> None:
+    """Ошибка валидации subject'а при заданных JWT/seed не печатает их (маркеры короткие, см. тест ниже)."""
+    jwt_mark, seed_mark = "JWTMARK", "SEEDMARK"
+    with pytest.raises(ValidationError) as exc_info:
+        settings_from_env(
+            OPENBAO__NATS__NATS_JWT=jwt_mark,
+            OPENBAO__NATS__NATS_SEED=seed_mark,
+            OPENBAO__NATS__NATS_RESULT_SUBJECT="files.*",
+        )
+    assert "NATS_RESULT_SUBJECT" in str(exc_info.value)
+    assert jwt_mark not in str(exc_info.value)
+    assert seed_mark not in str(exc_info.value)
+
+
+def test_nats_secrets_hidden_in_repr(settings_from_env) -> None:
+    settings = settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED)
+    assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == JWT
+    assert JWT not in repr(settings) and SEED not in repr(settings)
+    assert JWT not in str(settings) and SEED not in str(settings)
+
+
+MINIO_KEYS = ("MINIO_URL", "MINIO_ADMIN_LOGIN", "MINIO_ADMIN_PASSWORD", "MINIO_BUCKET", "MINIO_SECURE")
+
+
+@pytest.mark.parametrize(
+    "missing",
+    [
+        ("OPENBAO__NATS__NATS_URL",),  # нет поля в группе NATS: в ошибке — вход группы NATS (с JWT)
+        tuple(f"OPENBAO__MINIO__{key}" for key in MINIO_KEYS),  # нет группы MINIO: в ошибке — вход всего OPENBAO
+    ],
+)
+def test_validation_error_hides_nats_secrets(settings_from_env, missing: tuple[str, ...]) -> None:
+    """Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
+
+    Маркеры короткие: pydantic обрезает `input_value` в тексте ошибки (около 50 символов по середине), и полный
+    JWT там не виден даже без скрытия — проверка на полном значении прошла бы вхолостую.
+    """
+    jwt_mark, seed_mark = "JWTMARK", "SEEDMARK"
+    with pytest.raises(ValidationError) as exc_info:
+        settings_from_env(
+            OPENBAO__NATS__NATS_JWT=jwt_mark, OPENBAO__NATS__NATS_SEED=seed_mark, **dict.fromkeys(missing)
+        )
+    assert jwt_mark not in str(exc_info.value)
+    assert seed_mark not in str(exc_info.value)

api/tests/test_contract.py

@@ -626,6 +626,12 @@ def test_request_subject_by_pipeline() -> None:
     assert NATS_MSG_ID_HEADER == "Nats-Msg-Id"
 
 
+def test_request_subject_uses_environment_map() -> None:
+    subjects = {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
+    assert request_subject("tabular", subjects) == "test.files.tabular_uploaded"
+    assert request_subject("image", subjects) == "test.files.img_uploaded"
+
+
 def test_request_subject_rejects_unknown_pipeline() -> None:
     with pytest.raises(ValueError, match="pipeline"):
         request_subject("ocr")

api/tests/test_nats_core.py

@@ -25,6 +25,7 @@ from src.core.nats import (
     NatsNotConnectedError,
     NatsPublishError,
     build_nats_auth,
+    used_subjects,
 )
 
 SRC_DIR = Path(__file__).resolve().parent.parent / "src"
@@ -377,3 +378,33 @@ def test_src_does_not_use_jetstream():
                     if alias.name.startswith("nats.js")
                 )
     assert offenders == []
+
+
+def test_used_subjects_core() -> None:
+    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
+    assert used_subjects(nats) == {
+        "tabular_request": "files.tabular_uploaded",
+        "image_request": "files.img_uploaded",
+        "result": "test.files.tabular_parsed",
+    }
+
+
+def test_connection_subscribes_to_configured_result_subject() -> None:
+    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
+    assert NatsConnection(nats).result_subject == "test.files.tabular_parsed"
+    assert NatsConnection(nats, result_subject="override").result_subject == "override"
+
+
+async def test_connection_subscribes_with_configured_result_subject() -> None:
+    """T3.2: subscribe вызывается с subject'ом из настроек (без override) и группой очереди."""
+    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed")
+    world = _FakeWorld()
+    conn = NatsConnection(nats, client_factory=world, sleep=_RecordingSleep(), queue="g")
+    await conn.start()
+    assert await conn.wait_ready(timeout=2)
+
+    await conn.subscribe_results(_handler)
+
+    subscription = world.clients[-1].subscriptions[0]
+    assert (subscription["subject"], subscription["queue"]) == ("test.files.tabular_parsed", "g")
+    await conn.stop()

api/tests/test_recognition_exchange.py

@@ -75,7 +75,8 @@ class Clock:
 class RequestSubscriber:
     """Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`."""
 
-    def __init__(self) -> None:
+    def __init__(self, subjects: tuple[str, ...] = REQUEST_SUBJECTS) -> None:
+        self.subjects = subjects
         self.messages: list[Msg] = []
         self._nc = NATS()
 
@@ -89,7 +90,7 @@ class RequestSubscriber:
         async def collect(msg: Msg) -> None:
             self.messages.append(msg)
 
-        for subject in REQUEST_SUBJECTS:
+        for subject in self.subjects:
             await self._nc.subscribe(subject, cb=collect)
         await self._nc.flush()
 
@@ -261,6 +262,67 @@ async def test_publish_request_to_service(  # noqa: PLR0913,PLR0917 - фикст
     assert job.attempt == 1
 
 
+@pytest.mark.integration
+async def test_publish_uses_configured_request_subject(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
+) -> None:
+    """Review Focus 4: subject — из настроек окружения, `request_subject` задания — фактический subject публикации."""
+    run = uuid.uuid4().hex[:8]
+    subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
+    custom = RequestSubscriber(subjects=tuple(subjects.values()))
+    await custom.start()
+    try:
+        job_id = await jobs.make_queued_job(
+            pipeline=RecognitionPipeline.tabular, file_type=FileType.XLSX, original_name="ОСВ 60.xlsx"
+        )
+        await publish_pending(
+            session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, request_subjects=subjects
+        )
+        (msg,) = await custom.wait_for(job_id, 1)
+    finally:
+        await custom.stop()
+    assert msg.subject == subjects["tabular"]
+    assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
+
+
+@pytest.mark.integration
+async def test_republish_updates_request_subject(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
+) -> None:
+    """I-1: повторная публикация уходит на актуальный subject и обновляет `request_subject` задания."""
+    run = uuid.uuid4().hex[:8]
+    subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
+    clock = Clock()
+    job_id = await jobs.make_queued_job(
+        pipeline=RecognitionPipeline.tabular,
+        file_type=FileType.XLSX,
+        original_name="ОСВ 61.xlsx",
+        reply_deadline=clock() + timedelta(hours=1),
+    )
+    assert (await _publish(session_factory, nats, clock)).published == 1
+    old_subject = (await jobs.get(job_id)).request_subject
+    assert old_subject != subjects["tabular"]
+
+    custom = RequestSubscriber(subjects=tuple(subjects.values()))
+    await custom.start()
+    try:
+        clock.advance(ACK + timedelta(seconds=1))
+        report = await publish_pending(
+            session_factory,
+            nats,
+            ack_timeout=ACK,
+            max_republish=MAX_REPUBLISH,
+            now=clock,
+            request_subjects=subjects,
+        )
+        assert report.republished == 1
+        (msg,) = await custom.wait_for(job_id, 1)
+    finally:
+        await custom.stop()
+    assert msg.subject == subjects["tabular"]
+    assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
+
+
 @pytest.mark.integration
 async def test_publish_keeps_existing_reply_deadline(
     session_factory: async_sessionmaker[AsyncSession],
@@ -819,7 +881,12 @@ async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.Monke
     await _wait_until(lambda: len(started) == EXCHANGE_TASKS)
 
     assert len(exchange.tasks) == EXCHANGE_TASKS
-    assert started["publisher"] == {"poll_seconds": 7, "ack_timeout": timedelta(seconds=11), "max_republish": 2}
+    assert started["publisher"] == {
+        "poll_seconds": 7,
+        "ack_timeout": timedelta(seconds=11),
+        "max_republish": 2,
+        "request_subjects": {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"},
+    }
     assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS}
     await exchange.stop()
     assert all(task.done() for task in exchange.tasks)

openspec/changes/nats-env-config/.openspec.yaml

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

openspec/changes/nats-env-config/design.md

@@ -0,0 +1,20 @@
+## Context
+
+Дизайн сквозной: `../docs/openspec/changes/nats-env-config/design.md` (решения 1–7, R-1…R-6). Здесь — только то, что
+специфично для `arm-core`.
+
+## Decisions
+
+1. Ядро публикует запросы в `tabular_request` и `image_request`, подписан на `result` (queue group `arm-core`); в `/healthcheck.subjects` — `tabular_request`, `image_request`, `result`.
+2. Проверяются все три subject'а, включая неиспользуемые сервисом (docs design, решение 1 и R-3).
+3. Выбор subject'а и подписки получают имена из `settings.OPENBAO.NATS`; константы `contract/subjects.py` остаются
+   умолчаниями полей настроек. Форма передачи (аргумент, объект subject'ов) — в плане.
+
+## Risks / Trade-offs
+
+См. docs design: разъехавшиеся настройки — тишина на шине; JWT без прав на заданные subject'ы — ошибка прав NATS в логе.
+
+## Decisions made without discussion
+
+Решения без обсуждения ведутся в сквозном `../docs/openspec/changes/nats-env-config/design.md` (R-1…R-6); решения,
+принятые при исполнении в этом репозитории, дописываются сюда с маркером `R-n` сквозной нумерации.

openspec/changes/nats-env-config/proposal.md

@@ -0,0 +1,40 @@
+## Why
+
+Прод и тест будут работать с разными наборами subject'ов NATS и разными JWT; платформа передаёт в под только
+`OPENBAO__*`. Сейчас subject'ы — константы `api/src/contract/subjects.py`. Решение владельца 12.10.2026: subject'ы и
+JWT задаются по окружению секретами группы `OPENBAO__NATS__`, как в `RSHB/armka_file_worker`. Сквозной change —
+`../docs/openspec/changes/nats-env-config` (RSHB/arm-ka-docs#80); здесь — его часть для `arm-core` (RSHB/arm-core#106).
+
+Срез: вне плана срезов `09-delivery-plan.md` — подготовка к контуру. Таблицы DBML v5.17 не затрагиваются
+(`recognition_job.request_subject` по-прежнему хранит фактический subject публикации).
+
+## What Changes
+
+- `NATSSettings`: `NATS_TABULAR_REQUEST_SUBJECT` (`files.tabular_uploaded`), `NATS_IMAGE_REQUEST_SUBJECT`
+  (`files.img_uploaded`), `NATS_RESULT_SUBJECT` (`files.tabular_parsed`); проверка всех трёх при старте (непустые, без
+  пробелов, `*`, `>` и пустых токенов, попарно различны).
+- `NATS_JWT`, `NATS_SEED` — `SecretStr`; `Settings(hide_input_in_errors=True)`.
+- Ядро публикует запросы в `tabular_request` и `image_request`, подписан на `result` (queue group `arm-core`) — по настройкам; константы `contract/subjects.py` — только умолчания.
+- Лог при старте и `/healthcheck.subjects` (`tabular_request`, `image_request`, `result`), независимо от состояния NATS.
+- e2e (`tests/e2e/`) остаётся на subject'ах по умолчанию: ему нужен контейнер `tabular-parser` стенда (docs design R-7).
+- `.env.example`, README.
+
+## Non-goals
+
+- `schema_version` контракта и копии схем не меняются (кроме побайтной синхронизации `description` из docs, если она
+  изменится).
+- Queue group'ы — константы; механизм nkeys/JWT не меняется.
+- Пароли MinIO/SMB в `SecretStr` не переводятся.
+- Настройка через `OPENBAO__APP__*` не вводится.
+
+## Capabilities
+
+### New Capabilities
+
+### Modified Capabilities
+- `recognition-job-exchange`: subject'ы обмена и учётные данные NATS — настройки окружения; `/healthcheck.subjects`.
+
+## Impact
+
+Файлы: `api/src/core/config.py`, `api/src/contract/subjects.py`, `api/src/core/nats.py`, `api/src/services/recognition_exchange.py`, `api/src/services/primary_files.py`, `api/src/app.py` (`/healthcheck`, лог старта), тесты (`tests/test_config*.py`, `tests/test_nats_*.py`, `tests/e2e/`), `api/.env.example`, README. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём
+сервисам.

openspec/changes/nats-env-config/specs/recognition-job-exchange/spec.md

@@ -0,0 +1,46 @@
+## ADDED Requirements
+
+### Requirement: Subject'ы и учётные данные NATS из настроек окружения
+Ядро SHALL брать subject'ы обмена из `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT`, `NATS_IMAGE_REQUEST_SUBJECT`,
+`NATS_RESULT_SUBJECT` (умолчания `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed`): публикует запросы в `tabular_request` и `image_request`, подписан на `result` (queue group `arm-core`).
+При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
+попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
+логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
+`/healthcheck` объектом `subjects` с ключами `tabular_request`, `image_request`, `result` при любом значении `nats` (`ok | down | off`).
+
+#### Scenario: Умолчания
+- **WHEN** приложение стартует без ключей `OPENBAO__NATS__NATS_*_SUBJECT`
+- **THEN** `/healthcheck.subjects` содержит имена `files.*` по умолчанию, обмен идёт по ним
+
+#### Scenario: Subject'ы окружения
+- **WHEN** заданы `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=test.files.tabular_uploaded`,
+  `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=test.files.img_uploaded`,
+  `OPENBAO__NATS__NATS_RESULT_SUBJECT=test.files.tabular_parsed`
+- **THEN** подписки и публикации идут только по этим subject'ам, `/healthcheck.subjects` их показывает
+
+#### Scenario: Недопустимый subject
+- **WHEN** `OPENBAO__NATS__NATS_RESULT_SUBJECT` совпадает с subject'ом запроса или содержит `>`
+- **THEN** приложение не стартует, ошибка называет ключ
+
+#### Scenario: JWT не раскрывается
+- **WHEN** заданы `NATS_JWT` и `NATS_SEED`, а валидация настроек падает на другом поле
+- **THEN** текст ошибки и лог не содержат значений JWT и seed; `repr` настроек показывает их скрытыми
+
+## MODIFIED Requirements
+
+### Requirement: Соединение с NATS по шаблону, без JetStream
+Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных,
+если они пусты; подписываться на subject ответа `NATS_RESULT_SUBJECT` (по умолчанию `files.tabular_parsed`) в 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`

openspec/changes/nats-env-config/tasks.md

@@ -0,0 +1,18 @@
+## 1. Тесты
+
+- [x] 1.1 Unit: умолчания subject'ов, чтение `OPENBAO__NATS__NATS_*_SUBJECT`, отказ на `*`/`>`/пробеле/пустом токене и
+  совпадающих именах; `SecretStr` для `NATS_JWT`/`NATS_SEED`, отсутствие значений в ошибке валидации и `repr`
+- [x] 1.2 Unit: `/healthcheck.subjects` (`tabular_request`, `image_request`, `result`) при `nats = ok | down | off`; лог старта с subject'ами
+- [x] 1.3 Интеграционный тест: приложение с subject'ами окружения публикует запрос и подписано на ответ по ним
+
+## 2. Реализация
+
+- [x] 2.1 `core/config.py`: поля и валидация `NATSSettings`, `SecretStr`, `hide_input_in_errors=True`
+- [x] 2.2 Колбэки nkeys — `get_secret_value()`
+- [x] 2.3 Подписка и публикация по настройкам; константы `contract/subjects.py` — умолчания
+- [x] 2.4 `/healthcheck.subjects`, лог при старте
+
+## 3. Документы
+
+- [x] 3.1 `api/.env.example`, README (переменные NATS, `/healthcheck`)
+- [x] 3.2 `make check-full` зелёный