README.md

@@ -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": "...", "image_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/src/app.py

@@ -4,8 +4,9 @@
 NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `files.tabular_parsed`
 в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается),
 middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет РАИСА лишь условно импортируется,
-middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API
-(`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`.
+middleware не подключается), healthcheck (uptime, version, nats: ok|down|off, 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 подключено, цикл маппинга результатов
@@ -43,7 +44,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
@@ -183,6 +184,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 +243,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/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,6 +61,15 @@ 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`.
 
@@ -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

@@ -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(
@@ -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",
         )

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

@@ -47,6 +47,25 @@ 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()
+
+
 async def test_healthcheck_nats_down_when_enabled_without_connection(
     client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
 ):

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,18 @@ 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"

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,29 @@ 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_publish_keeps_existing_reply_deadline(
     session_factory: async_sessionmaker[AsyncSession],
@@ -819,7 +843,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)