CLAUDE.md

@@ -4,7 +4,7 @@
 
 ## Что это
 Stateless-сервис разбора XLS/XLSX/XLSB: ОСВ по счетам (§1 правил) и ББ/ОФР из Excel (§12). Вход — NATS
-`files.tabular_uploaded` + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
+`files.tabular_uploaded` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT) + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
 в S3 `results/` + NATS `files.tabular_parsed`. **Нет доступа к БД.** Не знает про организации, периметры, план счетов —
 возвращает структуру файла; нормализация и резолв — в arm-core.
 

README.md

@@ -382,7 +382,7 @@ cd api && .venv/bin/python run_server.py
 curl -s localhost:8000/healthcheck
 ```
 
-В ответе — `uptime`, `version` и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет,
+В ответе — `uptime`, `version`, `subjects` (используемые subject'ы: `tabular_request`, `result`; JWT там нет) и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет,
 `off` — выключен (`NATS_ENABLED=false`, воркер не запущен).
 
 Если NATS или MinIO недоступны при старте, приложение всё равно поднимается: ошибка логируется, соединение с NATS
@@ -468,6 +468,9 @@ ENV_FILE=.env.test .venv/bin/python -m src.parser.cli <файл> [--file-type XL
 | `OPENBAO__NATS__NATS_URL` | — (обязательна) | адрес NATS, `nats://host:4222` |
 | `OPENBAO__NATS__NATS_SEED` | пусто | nkeys seed; пусто вместе с JWT — подключение без аутентификации (dev) |
 | `OPENBAO__NATS__NATS_JWT` | пусто | JWT пользователя NATS |
+| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса `tabular`, на который подписан сервис |
+| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса `image` (слушает scan-gateway; здесь — единый набор ключей) |
+| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject, в который сервис публикует ack и результат |
 | `DEBUG` | `false` | уровень логов DEBUG |
 | `ENV_FILE` | `.env` | путь к файлу окружения (от `api/`) |
 | `NATS_ENABLED` | `true` | `false` — не подключаться к NATS, воркер не стартует |

api/.env.example

@@ -27,6 +27,11 @@ OPENBAO__MINIO__MINIO_SECURE=false
 OPENBAO__NATS__NATS_URL=nats://localhost:4222
 OPENBAO__NATS__NATS_SEED=
 OPENBAO__NATS__NATS_JWT=
+# Subject'ы (необязательно; значения ниже — по умолчанию). Запрос tabular — подписка, результат — публикация.
+# Image-запрос слушает scan-gateway; здесь ключ нужен только для единого набора OPENBAO__NATS__*
+# 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
 
 # Необязательные настройки (значения ниже — значения по умолчанию из src/core/config.py)
 # Префикс путей приложения за прокси и версия в OpenAPI

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

@@ -2,7 +2,7 @@
 
 Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без
 аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на
-`files.tabular_uploaded` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
+subject запроса `tabular` из настроек после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
 (`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4).
 
 При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и
@@ -24,7 +24,7 @@ from src.core.clients import S3Storage
 from src.core.config import get_settings
 from src.core.exception_handlers import register_exception_handlers
 from src.core.logger import setup_logger
-from src.core.nats import NatsConnection
+from src.core.nats import NatsConnection, used_subjects
 from src.core.worker import STOP_TIMEOUT, RecognitionWorker
 from src.parser import get_parser
 
@@ -85,12 +85,14 @@ def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker
         service_version=VERSION,
         concurrency=settings.WORKER_CONCURRENCY,
         done_cache_size=settings.DONE_CACHE_SIZE,
+        result_subject=settings.OPENBAO.NATS.NATS_RESULT_SUBJECT,
     )
     return nats, worker
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+    logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS))
     storage = await _connect_s3()
     nats: NatsConnection | None = None
     worker: RecognitionWorker | None = None
@@ -130,7 +132,12 @@ def _nats_state(application: FastAPI) -> str:
 
 @app.get("/healthcheck", status_code=200)
 async def healthcheck(request: Request):
-    return {"uptime": UP_TIME, "version": VERSION, "nats": _nats_state(request.app)}
+    return {
+        "uptime": UP_TIME,
+        "version": VERSION,
+        "nats": _nats_state(request.app),
+        "subjects": used_subjects(settings.OPENBAO.NATS),
+    }
 
 
 app.include_router(router)

api/src/contract/subjects.py

@@ -26,10 +26,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

@@ -1,13 +1,21 @@
 import os
+import re
 from typing import Any, Literal
 
-from pydantic import NonNegativeInt, PositiveInt, model_validator
+from pydantic import NonNegativeInt, PositiveInt, SecretStr, ValidationInfo, field_validator, model_validator
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
 
+from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT
 from src.core.settings_sources import OpenBaoAppSettingsSource
 
+# 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 MinioSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     MINIO_URL: str
     MINIO_ADMIN_LOGIN: str
     MINIO_ADMIN_PASSWORD: str
@@ -16,12 +24,54 @@ 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_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 OpenBAOSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     MINIO: MinioSettings
     NATS: NATSSettings
 
@@ -37,6 +87,8 @@ class OpenBAOSettings(BaseSettings):
 
 
 class ServerSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
     HOST: str = "0.0.0.0"
     PORT: int = 8000
 
@@ -84,6 +136,7 @@ class Settings(BaseSettings):
         )
 
     model_config = SettingsConfigDict(
+        hide_input_in_errors=True,
         extra="ignore",
         env_file=os.environ.get("ENV_FILE", ".env"),
         env_nested_delimiter="__",

api/src/core/nats.py

@@ -10,7 +10,8 @@
   завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз
   вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и
   вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам.
-- `subscribe_requests(cb)` — подписка на `files.tabular_uploaded` в queue group `tabular-parser`: из нескольких
+- `subscribe_requests(cb)` — подписка на subject запроса `tabular` из настроек (по умолчанию
+  `files.tabular_uploaded`) в queue group `tabular-parser`: из нескольких
   инстанций сообщение получает одна. После `subscribe` — `flush`: сервер обработал SUB, ошибки соединения проявляются
   здесь. Отказ в правах (`Permissions Violation`) nats-py 2.14 так не проявляет: соединение остаётся открытым,
   `subscribe`/`flush`/`publish` проходят, сервер лишь присылает `-ERR` в `error_cb` — он логируется отдельно уровнем
@@ -37,7 +38,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, TABULAR_REQUEST_SUBJECT
+from src.contract.subjects import NATS_MSG_ID_HEADER
 from src.core.config import NATSSettings
 
 REQUEST_QUEUE_GROUP = "tabular-parser"
@@ -88,8 +89,8 @@ def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
     Пустые `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:
@@ -112,6 +113,11 @@ def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
     return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb}
 
 
+def used_subjects(settings: NATSSettings) -> dict[str, str]:
+    """Subject'ы сервиса для лога старта и `/healthcheck.subjects`: подписан на запрос `tabular`, публикует ответ."""
+    return {"tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT, "result": settings.NATS_RESULT_SUBJECT}
+
+
 class NatsConnection:
     """Соединение сервиса с NATS: фоновое подключение, подписка на запросы, публикация с повторами.
 
@@ -276,9 +282,13 @@ class NatsConnection:
 
     # --- подписка и публикация ------------------------------------------------------------------
     async def subscribe_requests(
-        self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT
+        self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str | None = None
     ) -> Any:
-        """Подписка на запросы (`files.tabular_uploaded`) в queue group; исключения `cb` логируются."""
+        """Подписка на запросы в queue group; исключения `cb` логируются.
+
+        Без `subject` — `NATS_TABULAR_REQUEST_SUBJECT` настроек (по умолчанию `files.tabular_uploaded`).
+        """
+        subject = subject if subject is not None else self._settings.NATS_TABULAR_REQUEST_SUBJECT
         if not self.is_connected:
             raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна")
 

api/src/core/worker.py

@@ -1,4 +1,5 @@
-"""Воркер распознавания: запрос `files.tabular_uploaded` → ack → S3 → парсер → S3 → `files.tabular_parsed`.
+"""Воркер распознавания: запрос на subject запросов → ack → S3 → парсер → S3 → subject ответа
+(по умолчанию `files.tabular_uploaded` / `files.tabular_parsed`).
 
 Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтверждение запроса до обработки», «Результат по
 контракту», «Ошибки транспорта превращаются в результат», «Парсер — заменяемый интерфейс».
@@ -141,11 +142,13 @@ class RecognitionWorker:
         concurrency: int = 1,
         done_cache_size: int = 1000,
         clock: Clock = time.monotonic,
+        result_subject: str = RESULT_SUBJECT,
     ) -> None:
         if concurrency < 1 or done_cache_size < 1:
             raise ValueError("concurrency и done_cache_size должны быть ≥ 1")
         self._storage = storage
         self._publisher = publisher
+        self._result_subject = result_subject
         self._parser = parser
         self._service = ServiceInfo.model_validate({"name": service_name, "version": service_version})
         self._concurrency = concurrency
@@ -198,7 +201,8 @@ class RecognitionWorker:
 
     # --- приём запроса ----------------------------------------------------------------------------------------------
     async def on_request(self, msg: InboundMessage) -> None:
-        """Обработчик подписки на `files.tabular_uploaded`; исключения наружу не выходят."""
+        """Обработчик подписки на subject запросов (по умолчанию `files.tabular_uploaded`);
+        исключения наружу не выходят."""
         if not self._accepting:
             logger.warning("Воркер останавливается: запрос не принят (без ack, ядро переотправит)")
             return
@@ -464,7 +468,7 @@ class RecognitionWorker:
         """Публикация конверта; исчерпанные повторы и прочие сбои — error-лог, не исключение (D-5)."""
         payload = json.dumps(envelope.to_wire(), ensure_ascii=False).encode("utf-8")
         try:
-            await self._publisher.publish(RESULT_SUBJECT, payload, msg_id)
+            await self._publisher.publish(self._result_subject, payload, msg_id)
         except NatsPublishError as exc:
             logger.error(
                 "Конверт не опубликован после всех повторов; ядро закроет задание по reply_deadline. "

api/tests/integration/test_roundtrip.py

@@ -3,8 +3,8 @@
 Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде:
 `_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с парсером по умолчанию
 (`get_parser(settings)` → `OsvParser`); исходник — битый zip, поэтому результат — ошибка `UNREADABLE`.
-Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `files.tabular_uploaded` с
+Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос на subject запросов прогона с
-`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `files.tabular_parsed`.
+`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты с subject'а ответа прогона.
 
 Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
 обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
@@ -12,8 +12,9 @@ Spec tabular-recognition-service: «Ack раньше результата», «
 «Реальная ОСВ через воркер» (маркер `fixtures`: реальный XLS из `../fixtures` → `done` и файл результата в MinIO),
 «Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 1.2`).
 
-Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
+Изоляция: subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83);
-работать другой сервис — сообщения `files.tabular_parsed` фильтруются по своему `job_id`. Ожидания — по событиям с
+`job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
+работать другой сервис — сообщения с subject'а ответа фильтруются по своему `job_id`. Ожидания — по событиям с
 таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок.
 """
 
@@ -37,8 +38,6 @@ from result_contract import schema_errors
 
 from src.contract import (
     NATS_MSG_ID_HEADER,
-    RESULT_SUBJECT,
-    TABULAR_REQUEST_SUBJECT,
     RecognitionResult,
     request_msg_id,
 )
@@ -71,6 +70,14 @@ SOURCE_BYTES = b"PK\x03\x04 round-trip: broken zip, the default parser answers U
 XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
 XLS_CONTENT_TYPE = "application/vnd.ms-excel"
 
+# Subject'ы прогона уникальны (OQ-83): контейнер tabular-parser стенда слушает умолчания и сообщения теста не получает
+RUN = uuid.uuid4().hex[:8]
+TEST_SUBJECTS = {
+    "NATS_TABULAR_REQUEST_SUBJECT": f"test-{RUN}.files.tabular_uploaded",
+    "NATS_IMAGE_REQUEST_SUBJECT": f"test-{RUN}.files.img_uploaded",
+    "NATS_RESULT_SUBJECT": f"test-{RUN}.files.tabular_parsed",
+}
+
 app_module = importlib.import_module("src.app")  # модуль, а не объект `app`, который реэкспортирует пакет `src`
 
 
@@ -117,7 +124,7 @@ def _assert_ack(msg: Msg, request: dict[str, Any]) -> None:
 
 
 class Core:
-    """Тестовый клиент в роли ядра: публикует запросы и собирает `files.tabular_parsed` по своим `job_id`."""
+    """Тестовый клиент в роли ядра: публикует запросы и собирает сообщения с subject'а ответа по своим `job_id`."""
 
     def __init__(self, client: NATS) -> None:
         self.client = client
@@ -136,7 +143,7 @@ class Core:
         self._inboxes.setdefault(request["job_id"], asyncio.Queue())
         payload = json.dumps(request, ensure_ascii=False).encode("utf-8")
         headers = {NATS_MSG_ID_HEADER: request_msg_id(request["job_id"], request["attempt"])}
-        await self.client.publish(TABULAR_REQUEST_SUBJECT, payload, headers=headers)
+        await self.client.publish(TEST_SUBJECTS["NATS_TABULAR_REQUEST_SUBJECT"], payload, headers=headers)
         await self.client.flush()
 
     async def next_message(self, request: dict[str, Any]) -> Msg:
@@ -144,7 +151,7 @@ class Core:
             return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT)
         except TimeoutError:
             pytest.fail(
-                f"нет сообщения files.tabular_parsed для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
+                f"нет сообщения на subject ответа прогона для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
                 "(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)",
                 pytrace=False,
             )
@@ -216,6 +223,8 @@ async def uploads() -> AsyncIterator[Uploads]:
 async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
     """Приложение в процессе теста: lifespan с `NATS_ENABLED=True` (в `.env.test` — false); выход — остановка."""
     monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
+    nats_settings = app_module.settings.OPENBAO.NATS
+    monkeypatch.setattr(app_module.settings.OPENBAO, "NATS", nats_settings.model_copy(update=TEST_SUBJECTS))
     application = app_module.app
     tasks: list[asyncio.Task[None]] = []
     async with application.router.lifespan_context(application):
@@ -232,11 +241,11 @@ async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
 
 @pytest.fixture
 async def core(service: None) -> AsyncIterator[Core]:
-    """Клиент «ядра», подписанный на `files.tabular_parsed` до первой публикации запроса."""
+    """Клиент «ядра», подписанный на subject ответа прогона до первой публикации запроса."""
     client = NATS()
     await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
     core = Core(client)
-    await client.subscribe(RESULT_SUBJECT, cb=core.on_result)
+    await client.subscribe(TEST_SUBJECTS["NATS_RESULT_SUBJECT"], cb=core.on_result)
     await client.flush()
     try:
         yield core

api/tests/unit/test_app.py

@@ -16,6 +16,7 @@ from typing import Any
 import httpx
 import pytest
 from loguru import logger
+from pydantic import SecretStr
 
 from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT
 from src.core.clients import S3Storage
@@ -124,6 +125,11 @@ def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[
     yield app_module
 
 
+def _expected_subjects(module: Any) -> dict[str, str]:
+    nats = module.settings.OPENBAO.NATS
+    return {"tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT, "result": nats.NATS_RESULT_SUBJECT}
+
+
 async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off(
     patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str]
 ) -> None:
@@ -136,7 +142,8 @@ async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off(
 
     assert response.status_code == 200  # noqa: PLR2004 - HTTP 200
     body = response.json()
-    assert set(body) == {"uptime", "version", "nats"}
+    assert set(body) == {"uptime", "version", "nats", "subjects"}
+    assert body["subjects"] == _expected_subjects(patched_app)
     assert body["nats"] == "off"
     assert body["version"] == patched_app.settings.VERSION
     assert FakeNatsConnection.instances == []
@@ -154,7 +161,7 @@ async def test_minimal_environment_starts_and_healthcheck_answers(
         response = await client.get("/healthcheck")
 
     assert response.status_code == 200  # noqa: PLR2004 - HTTP 200
-    assert set(response.json()) == {"uptime", "version", "nats"}
+    assert set(response.json()) == {"uptime", "version", "nats", "subjects"}
 
 
 async def test_nats_enabled_subscribes_worker_and_reports_ok(
@@ -195,12 +202,79 @@ async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, mon
     FakeNatsConnection.connect_on_start = False
 
     async with running_app(patched_app) as client:
-        assert (await client.get("/healthcheck")).json()["nats"] == "down"
+        response = await client.get("/healthcheck")
+        assert response.json()["nats"] == "down"
+        assert response.json()["subjects"] == _expected_subjects(patched_app)
+        assert "jwt" not in response.text.lower()
         (connection,) = FakeNatsConnection.instances
         assert connection.handlers == []  # подписка — только после подключения
         assert patched_app.app.state.worker is not None
 
 
+# --- subject'ы из настроек, JWT вне ответа, лог старта, передача result_subject воркеру ------------------------------
+CUSTOM_SUBJECTS = {
+    "NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded",
+    "NATS_RESULT_SUBJECT": "test.files.tabular_parsed",
+}
+SECRET_JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.healthcheck-secret.sig"
+
+
+@pytest.fixture
+def custom_nats_settings(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> Any:
+    """Подменяет `settings.OPENBAO.NATS` копией с нестандартными subject'ами и непустым JWT."""
+    nats = patched_app.settings.OPENBAO.NATS.model_copy(update={**CUSTOM_SUBJECTS, "NATS_JWT": SecretStr(SECRET_JWT)})
+    monkeypatch.setattr(patched_app.settings.OPENBAO, "NATS", nats)
+    return nats
+
+
+@pytest.mark.parametrize(("nats_enabled", "state"), [(False, "off"), (True, "down")])
+async def test_healthcheck_reports_configured_subjects_and_hides_jwt(
+    patched_app: Any,
+    custom_nats_settings: Any,
+    monkeypatch: pytest.MonkeyPatch,
+    nats_enabled: bool,
+    state: str,
+) -> None:
+    monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", nats_enabled)
+    FakeNatsConnection.connect_on_start = False  # state=down: NATS включён, но связи нет
+
+    async with running_app(patched_app) as client:
+        response = await client.get("/healthcheck")
+
+    body = response.json()
+    assert body["nats"] == state
+    assert body["subjects"] == {
+        "tabular_request": "test.files.tabular_uploaded",
+        "result": "test.files.tabular_parsed",
+    }
+    assert SECRET_JWT not in response.text
+    assert SECRET_JWT.split(".")[1] not in response.text
+
+
+async def test_startup_logs_configured_subjects(
+    patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch, log_records: list[Any]
+) -> None:
+    monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False)
+
+    async with running_app(patched_app):
+        pass
+
+    messages = [record["message"] for record in log_records if "Subject'ы NATS" in record["message"]]
+    assert len(messages) == 1
+    assert "test.files.tabular_uploaded" in messages[0]
+    assert "test.files.tabular_parsed" in messages[0]
+    assert SECRET_JWT not in messages[0]
+
+
+async def test_build_worker_passes_configured_result_subject(
+    patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Воркер публикует ack и результат в настроенный `NATS_RESULT_SUBJECT`, а не в константу контракта."""
+    nats, worker = patched_app._build_worker(FakeStorage([]))
+    assert nats.settings.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
+    assert worker._result_subject == "test.files.tabular_parsed"  # noqa: SLF001 - проверяем проводку настроек
+
+
 # --- проверка бакета при старте (_connect_s3) ------------------------------------------------------------------------
 @pytest.fixture
 def log_records() -> Iterator[list[Any]]:

api/tests/unit/test_config.py

@@ -8,6 +8,42 @@ from pydantic import ValidationError
 from src.core.config import Settings
 
 SettingsFactory = Callable[..., Settings]
+SettingsFromEnvFactory = Callable[..., Settings]
+
+# Минимальный набор обязательных переменных (как MINIMAL_ENV в conftest): окружение процесса, без .env
+MINIMAL_ENV_VARS = {
+    "DEBUG": "true",
+    "OPENBAO__MINIO__MINIO_URL": "localhost:9000",
+    "OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "minioadmin",
+    "OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "minioadmin",
+    "OPENBAO__MINIO__MINIO_BUCKET": "arm-ka",
+    "OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
+}
+
+
+@pytest.fixture
+def settings_from_env(minimal_settings: SettingsFactory, monkeypatch: pytest.MonkeyPatch) -> SettingsFromEnvFactory:
+    """Фабрика `settings_from_env(**env)`: Settings из переменных окружения поверх минимального набора.
+
+    Окружение процесса очищено фикстурой `minimal_settings` (она же убирает `OPENBAO__*`); значение `None` — удалить
+    переменную. `.env` не читается (`_env_file=None`).
+    """
+
+    def build(**env: str | None) -> Settings:
+        for name, value in MINIMAL_ENV_VARS.items():
+            monkeypatch.setenv(name, value)
+        for name, value in env.items():
+            if value is None:
+                monkeypatch.delenv(name, raising=False)
+            else:
+                monkeypatch.setenv(name, value)
+        return Settings(_env_file=None)  # type: ignore[call-arg]
+
+    return build
+
+
+JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
+SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
 
 
 def test_minimal_environment_gives_defaults(minimal_settings: SettingsFactory) -> None:
@@ -16,8 +52,8 @@ def test_minimal_environment_gives_defaults(minimal_settings: SettingsFactory) -
     assert settings.DEBUG is True
     assert settings.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
     assert settings.OPENBAO.NATS.NATS_URL == "nats://localhost:4222"
-    assert settings.OPENBAO.NATS.NATS_SEED == ""
+    assert settings.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
-    assert settings.OPENBAO.NATS.NATS_JWT == ""
+    assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
     assert settings.NATS_ENABLED is True
     assert settings.SERVICE_NAME == "tabular-parser"
     assert settings.WORKER_CONCURRENCY == 1
@@ -48,3 +84,93 @@ def test_service_name_other_than_contract_value_is_rejected(minimal_settings: Se
     """SERVICE_NAME — `service.name` конвертов контракта: другое значение — ошибка конфигурации при старте."""
     with pytest.raises(ValidationError, match="SERVICE_NAME"):
         minimal_settings("SERVICE_NAME=scan-gateway\n")
+
+
+def test_nats_subjects_default_to_contract_names(settings_from_env: SettingsFromEnvFactory) -> 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: SettingsFromEnvFactory) -> 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"
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
+)
+@pytest.mark.parametrize("key", ["NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT"])
+def test_invalid_subject_rejected(settings_from_env: SettingsFromEnvFactory, 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: SettingsFromEnvFactory, first: str, second: str) -> None:
+    """Любая пара из трёх subject'ов не может совпадать; сообщение называет оба ключа."""
+    with pytest.raises(ValidationError, match=f"{first} и {second} совпадают"):
+        settings_from_env(
+            **{f"OPENBAO__NATS__{first}": "test.same_subject", f"OPENBAO__NATS__{second}": "test.same_subject"}
+        )
+
+
+def test_nats_secrets_hidden_in_repr(settings_from_env: SettingsFromEnvFactory) -> 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: SettingsFromEnvFactory, missing: tuple[str, ...]
+) -> None:
+    """Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
+
+    pydantic в тексте ошибки усекает значения (`'SUACSSL3UA...PQYYYCU3VY'`), поэтому полный секрет там не появляется
+    даже без скрытия; проверяем края секретов — они и попадают в текст при утечке.
+    """
+    with pytest.raises(ValidationError) as exc_info:
+        settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
+    text = str(exc_info.value)
+    for secret in (JWT, SEED):
+        for fragment in (secret[:10], secret[-10:]):
+            assert fragment not in text
+
+
+def test_invalid_subject_error_hides_nats_secrets(settings_from_env: SettingsFromEnvFactory) -> None:
+    """Невалидный subject при заданных JWT/seed: ошибка валидации печатает вход группы NATS — секретов там нет."""
+    with pytest.raises(ValidationError) as exc_info:
+        settings_from_env(
+            OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, OPENBAO__NATS__NATS_RESULT_SUBJECT="files.*"
+        )
+    text = str(exc_info.value)
+    assert "NATS_RESULT_SUBJECT" in text
+    for secret in (JWT, SEED):
+        for fragment in (secret[:10], secret[-10:]):
+            assert fragment not in text

api/tests/unit/test_contract.py

@@ -650,6 +650,15 @@ def test_request_subject_rejects_unknown_pipeline() -> None:
         request_subject("ocr")
 
 
+def test_request_subject_uses_environment_map() -> None:
+    """Карта `subjects` окружения (NATSSettings.request_subjects()) подменяет имена контракта по умолчанию."""
+    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"
+    with pytest.raises(ValueError, match="неизвестный pipeline"):
+        request_subject("ocr", subjects)
+
+
 def test_request_subject_matches_example_pipeline() -> None:
     for name in ("request-tabular.json", "request-image.json"):
         request = RecognitionRequest.model_validate(_example(name))

api/tests/unit/test_nats.py

@@ -26,6 +26,7 @@ from src.core.nats import (
     NatsPublishError,
     Publisher,
     build_nats_auth,
+    used_subjects,
 )
 from src.core.worker import STOP_TIMEOUT
 
@@ -347,6 +348,26 @@ async def test_subscribe_requests_uses_tabular_subject_and_queue_group():
     await conn.stop()
 
 
+async def test_subscribe_requests_default_subject_comes_from_settings():
+    world = _FakeWorld()
+    settings = NATSSettings(
+        NATS_URL="nats://localhost:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded"
+    )
+    conn = NatsConnection(settings, client_factory=world, sleep=_RecordingSleep())
+    await conn.start()
+    assert await conn.wait_ready(1)
+
+    await conn.subscribe_requests(_noop_handler)
+
+    assert world.clients[-1].subscriptions[0]["subject"] == "test.files.tabular_uploaded"
+    await conn.stop()
+
+
+def test_used_subjects_tabular_parser() -> None:
+    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded")
+    assert used_subjects(nats) == {"tabular_request": "test.files.tabular_uploaded", "result": "files.tabular_parsed"}
+
+
 async def test_subscribe_requests_flushes_after_subscribe():
     """T8.3: подписка считается активной, когда сервер обработал SUB (flush после subscribe)."""
     world = _FakeWorld()

api/tests/unit/test_worker.py

@@ -174,7 +174,14 @@ def log_records() -> Iterator[list[Any]]:
 
 
 class Env:
-    def __init__(self, parser: Any = None, *, concurrency: int = 1, done_cache_size: int = 1000) -> None:
+    def __init__(
+        self,
+        parser: Any = None,
+        *,
+        concurrency: int = 1,
+        done_cache_size: int = 1000,
+        result_subject: str | None = None,
+    ) -> None:
         self.events: list[tuple[str, str]] = []
         self.storage = FakeStorage(self.events)
         self.storage.objects[(BUCKET, SOURCE_KEY)] = b"PK\x03\x04 xlsx bytes"
@@ -188,6 +195,7 @@ class Env:
             concurrency=concurrency,
             done_cache_size=done_cache_size,
             clock=FakeClock(),
+            **({} if result_subject is None else {"result_subject": result_subject}),
         )
 
     def bodies(self) -> list[dict[str, Any]]:
@@ -217,6 +225,20 @@ def _assert_message_valid(body: dict[str, Any]) -> None:
     RecognitionResultMessage.model_validate(body)
 
 
+async def test_ack_and_result_go_to_configured_result_subject() -> None:
+    env = await _make_env(result_subject="test.files.tabular_parsed")
+    try:
+        await env.worker.on_request(_msg(_request()))
+        await env.worker.join()
+    finally:
+        await env.worker.stop(timeout=1)
+
+    assert [(subject, msg_id) for subject, _body, msg_id in env.publisher.published] == [
+        ("test.files.tabular_parsed", f"ack:{JOB_ID}:1"),
+        ("test.files.tabular_parsed", f"result:{JOB_ID}:1"),
+    ]
+
+
 # --- ack раньше результата ------------------------------------------------------------------------------------------
 async def test_ack_is_published_before_s3_read_and_result(env: Env) -> None:
     await env.worker.on_request(_msg(_request()))

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). Здесь — только то, что
+специфично для `tabular-parser`.
+
+## Decisions
+
+1. Сервис подписан на `tabular_request` (queue group `tabular-parser`), публикует ack и результат в `result`; в `/healthcheck.subjects` — `tabular_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); здесь — его часть для `tabular-parser` (RSHB/arm-tabular-parser#31).
+
+Срез: вне плана срезов `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` (queue group `tabular-parser`), публикует ack и результат в `result` — по настройкам; константы `contract/subjects.py` — только умолчания.
+- Лог при старте и `/healthcheck.subjects` (`tabular_request`, `result`), независимо от состояния NATS.
+- Тесты roundtrip (`tests/integration/test_roundtrip.py`) задают собственные subject'ы через настройки (OQ-83).
+- `.env.example`, README.
+
+## Non-goals
+
+- `schema_version` контракта и копии схем не меняются (кроме побайтной синхронизации `description` из docs, если она
+  изменится).
+- Queue group'ы — константы; механизм nkeys/JWT не меняется.
+- Пароли MinIO/SMB в `SecretStr` не переводятся.
+- Настройка через `OPENBAO__APP__*` не вводится.
+
+## Capabilities
+
+### New Capabilities
+
+### Modified Capabilities
+- `tabular-recognition-service`: subject'ы обмена и учётные данные NATS — настройки окружения; `/healthcheck.subjects`.
+
+## Impact
+
+Файлы: `api/src/core/config.py`, `api/src/contract/subjects.py`, `api/src/core/nats.py`, `api/src/core/worker.py`, `api/src/app.py` (`/healthcheck`, лог старта), тесты (`tests/unit/test_config*.py`, `tests/unit/test_nats.py`, `tests/unit/test_app.py`, `tests/integration/test_roundtrip.py`), `api/.env.example`, README. Выкат обратно совместим (умолчания — текущие имена); набор окружения задаётся одинаково всем трём
+сервисам.

openspec/changes/nats-env-config/specs/tabular-recognition-service/spec.md

@@ -0,0 +1,43 @@
+## 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` (queue group `tabular-parser`), публикует ack и результат в `result`.
+При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
+попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
+логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
+`/healthcheck` объектом `subjects` с ключами `tabular_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` настроек показывает их скрытыми
+
+#### Scenario: Тесты на своих subject'ах
+- **WHEN** тесты roundtrip (`tests/integration/test_roundtrip.py`) запускаются при работающих контейнерах стенда на том же NATS
+- **THEN** тесты используют уникальные subject'ы из настроек и не получают и не забирают сообщения стенда
+
+## MODIFIED Requirements
+
+### Requirement: Соединение с NATS по правилам шаблона, без JetStream
+При пустых `NATS_SEED`/`NATS_JWT` сервис SHALL подключаться без учётных данных, иначе — с nkeys/JWT из `OPENBAO__NATS__NATS_SEED`/`NATS_JWT` (значения — секреты). Сервис MUST NOT
+требовать stream'ов или consumer'ов JetStream; при разрыве соединения SHALL переподключаться автоматически и восстанавливать
+подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`; `off` — NATS выключен флагом
+`NATS_ENABLED=false`).
+
+#### Scenario: NATS перезапущен
+- **WHEN** контейнер NATS перезапущен во время работы сервиса
+- **THEN** в течение 20 с после старта NATS подписка восстановлена, `nats = ok`, новый запрос обрабатывается

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`, `result`) при `nats = ok | down | off`; лог старта с subject'ами
+- [x] 1.3 Тесты roundtrip (`tests/integration/test_roundtrip.py`) — уникальные 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` зелёный