RSHB/arm-ka-docs
9
0
Fork 0
Code Issues 13 Pull Requests 2 Actions Packages Projects Releases Wiki Activity
arm-ka-docs/docs/superpowers/plans/2026-10-13-nats-env-config.md
Andrey Epifancev e197bd53e3 План nats-env-config: путь к venv сервиса 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 13:38:50 +04:00

50 KiB
Raw Blame History

Subject'ы и JWT NATS по окружению (OpenBao) — план реализации

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: subject'ы обмена NATS и учётные данные NATS задаются секретами OPENBAO__NATS__* по окружению (прод/тест) во всех трёх сервисах; subject'ы видны в логе старта и в /healthcheck.

Architecture: в NATSSettings каждого сервиса — три поля subject'ов с умолчаниями из констант contract/subjects.py, валидация при старте, SecretStr для JWT/seed, hide_input_in_errors во всех классах настроек. Код публикации и подписок берёт имена из настроек; функция used_subjects(nats_settings) в core/nats.py каждого сервиса — единый источник для лога и /healthcheck.subjects. Контракт (schema_version) не меняется.

Tech Stack: Python 3.11, pydantic v2 / pydantic-settings, nats-py (core NATS), FastAPI, pytest.

Spec: docs/openspec/changes/nats-env-config/ (proposal, design, specs/recognition-contract/spec.md) и дочерние change'и nats-env-config в arm-core/, tabular-parser/, scan-gateway/ (openspec/changes/nats-env-config/).

Рабочее место

Workspace-worktree: /Users/andrey.epifancev/projects/awp-ca/.wt/nats-env/ (дальше $W).

Репозиторий Путь Ветка Issue
docs $W/docs feat/80-nats-env-config RSHB/arm-ka-docs#80
arm-core $W/arm-core feat/106-nats-env-config RSHB/arm-core#106
tabular-parser $W/tabular-parser feat/31-nats-env-config RSHB/arm-tabular-parser#31
scan-gateway $W/scan-gateway feat/32-nats-env-config RSHB/arm-scan-gateway#32
  • git — только git -C <абсолютный путь worktree>; никогда не cd в основные checkout'ы awp-ca/<repo> (они на main, хук guard-main заблокирует всю сессию).
  • В каждом сервисном worktree перед первой задачей: make -C $W/<repo> venv (uv, Python 3.11).
  • Перед изменением кода — codegraph_explore с projectPath = абсолютный путь worktree сервиса.
  • Коммиты на русском, с трейлером Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>.
  • Инфраструктура для интеграционных тестов — compose-проект arm-ka (make -C $W/docs/dev up или уже поднятый стенд).

Global Constraints

  • Ключи (одинаковые в трёх сервисах): 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).
  • Каждый сервис проверяет все три имени: непустое, без пробельных символов, без * и >, без пустых токенов (a..b, ведущая/завершающая точка); три имени попарно различны; нарушение — сервис не стартует, ошибка называет ключ.
  • NATS_JWT, NATS_SEED — SecretStr; значения не попадают в логи, repr настроек и тексты ошибок валидации.
  • /healthcheck — новый объект subjects: arm-core {tabular_request, image_request, result}, tabular-parser {tabular_request, result}, scan-gateway {image_request, result}; отдаётся при любом состоянии NATS.
  • Queue group'ы (arm-core, tabular-parser, scan-gateway) — константы, не меняются.
  • schema_version контракта не меняется; копии схем в сервисах — побайтно равны docs/contracts/.
  • contract/subjects.py — одинаковый модуль в трёх репозиториях: менять одинаково.
  • Пароли MinIO/SMB в SecretStr не переводить (design R-5).
  • e2e ядра (tests/e2e/) остаётся на subject'ах по умолчанию (design R-7).
  • Язык комментариев, docstring'ов, сообщений об ошибках — русский; идентификаторы — английские.

Review Focus

  1. Утечка JWT через ошибку «группа не полная». Задан OPENBAO__NATS__NATS_JWT, но нет NATS_URL (или нет группы MINIO) → pydantic печатает input_value всей группы. Ожидание: ни JWT, ни seed в str(ValidationError). Тест — в задачах 2, 4, 6 (test_validation_error_hides_nats_secrets).
  2. Пробелы по краям значения из OpenBao ("files.x " — частая ошибка копирования в секрет). Ожидание: отказ старта с именем ключа, а не тихая подписка на несуществующий subject. Тест — параметр "files.tabular_uploaded " в test_invalid_subject_rejected (задачи 2, 4, 6).
  3. Совпадение subject'а ответа с subject'ом запроса — сервис получал бы собственные ответы. Ожидание: отказ старта, в сообщении оба ключа. Тест — test_equal_subjects_rejected (задачи 2, 4, 6).
  4. Ядро после смены секретов повторно публикует задание в новый subject, а recognition_job.request_subject отражает фактический subject публикации. Тест — test_publish_uses_configured_request_subject (задача 3).
  5. /healthcheck при выключенном или упавшем NATS всё равно показывает subjects и не показывает JWT. Тест — test_healthcheck_subjects_* (задачи 3, 5, 7).

Раздел A. Общий код настроек NATS (одинаковый в трёх сервисах)

Задачи 2, 4, 6 вставляют этот код дословно (отличаются только окружающие файлы).

A.1. Импорты в начало api/src/core/config.py (добавить к существующим; re — если ещё нет):

import re

from pydantic import SecretStr, ValidationInfo, field_validator, model_validator

from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT

A.2. Регулярное выражение (на уровне модуля, рядом с прочими константами модуля):

# 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")

A.3. Класс NATSSettings (заменяет существующий целиком):

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: SecretStr = SecretStr("")
    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}

A.4. hide_input_in_errors=True — в model_config каждого класса настроек config.py (MinioSettings, OpenBAOSettings, DBSettings/SMBSettings/ServerSettings — какие есть, и Settings). У классов без model_config добавить model_config = SettingsConfigDict(hide_input_in_errors=True); у Settings — дописать ключ в существующий SettingsConfigDict(...). Причина: ошибка «нет поля» печатает input_value всей группы, а в ней JWT.

A.5. contract/subjects.py — request_subject с картой subject'ов (заменить функцию; docstring модуля не трогать):

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

A.6. Колбэки nkeys — в build_nats_auth (api/src/core/nats.py) читать значения через get_secret_value(): seed = settings.NATS_SEED.get_secret_value() и jwt = settings.NATS_JWT.get_secret_value() (в scan-gateway сохранить существующий .strip(): settings.NATS_SEED.get_secret_value().strip()). Остальная логика функции не меняется.

A.7. Общие тесты настроек — добавить в тест-файл настроек сервиса (имя файла и фабрика настроек — в задаче). Фабрика make(**env) в каждом репозитории своя; в коде ниже она называется settings_from_env:

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"


@pytest.mark.parametrize(
    "value", ["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "]
)
def test_invalid_subject_rejected(settings_from_env, value: str) -> None:
    with pytest.raises(ValidationError, match="NATS_RESULT_SUBJECT"):
        settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT=value)


def test_equal_subjects_rejected(settings_from_env) -> None:
    with pytest.raises(ValidationError, match="NATS_TABULAR_REQUEST_SUBJECT и NATS_RESULT_SUBJECT совпадают"):
        settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT="files.tabular_uploaded")


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)."""
    with pytest.raises(ValidationError) as exc_info:
        settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
    assert JWT not in str(exc_info.value)
    assert SEED not in str(exc_info.value)

Проверка теста на честность (обязательна): временно убрать hide_input_in_errors из всех классов — оба параметра должны упасть (JWT виден в тексте ошибки). Если какой-то параметр проходит и без скрытия — тест не ловит утечку, доработать параметр под фактическое поведение pydantic, а не ослаблять проверку.

Семантика фабрики settings_from_env(**env): собирает Settings(_env_file=None) из минимального набора обязательных OPENBAO__* сервиса, поверх — env; значение None означает «удалить переменную». Фабрика строится на monkeypatch и существующих фикстурах тест-файла (см. задачу).

Task 1: docs — описания subject'ов в схемах контракта

Model: haiku (готовый текст).

Files:

  • Modify: $W/docs/contracts/recognition-request.schema.json (поле верхнего уровня description)
  • Modify: $W/docs/contracts/recognition-result-message.schema.json (поле верхнего уровня description)

Interfaces:

  • Produces: новые байты двух схем — задачи 2, 4, 6 копируют их в сервисы (api/contracts/).

  • Step 1: request. В description заменить начало Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). на Тело сообщения на subject запроса; subject — настройка окружения и должен соответствовать полю pipeline (tabular → NATS_TABULAR_REQUEST_SUBJECT, по умолчанию files.tabular_uploaded; image → NATS_IMAGE_REQUEST_SUBJECT, по умолчанию files.img_uploaded). Остальной текст description, $id, schema_version — без изменений.

  • Step 2: result-message. В description заменить Тело сообщения на subject files.tabular_parsed. на Тело сообщения на subject ответа (настройка окружения NATS_RESULT_SUBJECT, по умолчанию files.tabular_parsed).

  • Step 3: проверка. Run: make -C $W/docs check. Expected: зелёный (JSON валиден, примеры проходят схемы).

  • Step 4: Commit.

git -C $W/docs add contracts/recognition-request.schema.json contracts/recognition-result-message.schema.json
git -C $W/docs commit -m "contracts: subject'ы запроса и ответа — настройки окружения (nats-env-config)"

Task 2: arm-core — настройки NATS, контракт, nkeys

Model: haiku (код в разделе A).

Files:

  • Modify: $W/arm-core/api/src/core/config.py (A.1–A.4)
  • Modify: $W/arm-core/api/src/contract/subjects.py (A.5)
  • Modify: $W/arm-core/api/src/core/nats.py — build_nats_auth (A.6)
  • Modify: $W/arm-core/api/contracts/recognition-request.schema.json, recognition-result-message.schema.json (копия из $W/docs/contracts/, Task 1)
  • Test: $W/arm-core/api/tests/test_config.py, $W/arm-core/api/tests/test_contract.py, $W/arm-core/api/tests/test_nats_core.py

Interfaces:

  • Consumes: схемы из Task 1.

  • Produces: NATSSettings.NATS_TABULAR_REQUEST_SUBJECT/NATS_IMAGE_REQUEST_SUBJECT/NATS_RESULT_SUBJECT: str, NATSSettings.request_subjects() -> dict[str, str], NATS_SEED/NATS_JWT: SecretStr, request_subject(pipeline: str, subjects: Mapping[str, str] = REQUEST_SUBJECTS) -> str.

  • Step 1: фабрика и тесты. В tests/test_config.py добавить фикстуру на основе существующей clean_env и словаря required из test_defaults (вынести словарь в константу модуля REQUIRED_ENV):

@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

и тесты A.7 дословно. Существующие assert s.OPENBAO.NATS.NATS_SEED == "" / NATS_JWT == "" заменить на .get_secret_value() == "". В tests/test_contract.py добавить:

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"

В tests/test_nats_core.py хелпер NATSSettings(NATS_URL=..., NATS_SEED=seed, NATS_JWT=jwt) оставить (pydantic примет str в SecretStr).

  • Step 2: тесты падают. Run: cd $W/arm-core/api && .venv/bin/pytest tests/test_config.py tests/test_contract.py -q Expected: FAIL (NATS_RESULT_SUBJECT нет, request_subject не принимает второй аргумент).

  • Step 3: реализация. Применить A.1–A.6 к файлам задачи; скопировать схемы: cp $W/docs/contracts/recognition-request.schema.json $W/docs/contracts/recognition-result-message.schema.json $W/arm-core/api/contracts/.

  • Step 4: тесты проходят. Run: make -C $W/arm-core lint test Expected: PASS, ruff/mypy чистые.

  • Step 5: Commit.

git -C $W/arm-core add api/src/core/config.py api/src/contract/subjects.py api/src/core/nats.py api/contracts api/tests
git -C $W/arm-core commit -m "NATS: subject'ы и JWT из OPENBAO__NATS__*, SecretStr, скрытие входа в ошибках (#106)"

Task 3: arm-core — публикация, подписка, лог и /healthcheck по настройкам

Model: sonnet (интеграция в существующий код).

Files:

  • Modify: $W/arm-core/api/src/core/nats.py — NatsConnection.__init__, новая used_subjects
  • Modify: $W/arm-core/api/src/services/recognition_exchange.py — _process_one, publish_pending, run_publisher_loop, ExchangeTasks.start
  • Modify: $W/arm-core/api/src/services/primary_files.py:483 — request_subject при создании задания
  • Modify: $W/arm-core/api/src/app.py — lifespan (лог), healthcheck, docstring модуля
  • Modify: $W/arm-core/api/.env.example, $W/arm-core/README.md (раздел переменных NATS / healthcheck — найти grep'ом NATS_JWT и healthcheck)
  • Test: $W/arm-core/api/tests/test_nats_core.py, $W/arm-core/api/tests/test_recognition_exchange.py, $W/arm-core/api/tests/test_app.py, $W/arm-core/api/tests/test_api_files_upload.py

Interfaces:

  • Consumes: Task 2 (NATSSettings, request_subjects(), request_subject(pipeline, subjects)).

  • Produces: used_subjects(settings: NATSSettings) -> dict[str, str] в src/core/nats.py (ключи tabular_request, image_request, result); publish_pending(..., request_subjects: Mapping[str, str] = REQUEST_SUBJECTS); NatsConnection(settings, *, result_subject: str | None = None, ...) — None → settings.NATS_RESULT_SUBJECT.

  • Step 1: тесты.

    tests/test_nats_core.py:

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"

tests/test_recognition_exchange.py — по образцу существующего теста первой публикации (строки ~235–262: фикстуры jobs, RequestSubscriber, publish_pending), но RequestSubscriber подписывается на свои subject'ы: добавить в RequestSubscriber.__init__ параметр subjects: tuple[str, ...] = REQUEST_SUBJECTS и использовать его в start(); новый тест:

async def test_publish_uses_configured_request_subject(jobs: CommittedJobs, nats_conn) -> 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"}
    subscriber = RequestSubscriber(subjects=tuple(subjects.values()))
    await subscriber.start()
    try:
        job_id = await jobs.create(pipeline=RecognitionPipeline.tabular)
        await publish_pending(jobs.session_factory, nats_conn, ack_timeout=ACK, max_republish=MAX_REPUBLISH,
                              request_subjects=subjects)
        (msg,) = await subscriber.wait_for(job_id, 1)
    finally:
        await subscriber.stop()
    assert msg.subject == subjects["tabular"]
    assert (await jobs.get(job_id)).request_subject == subjects["tabular"]

Имена фикстур (jobs, соединения NATS, метода создания задания) взять из существующих тестов файла — если они называются иначе, использовать существующие, не создавая новых.

tests/test_app.py:

async def test_healthcheck_subjects_when_nats_off(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
    body = (await client.get("/healthcheck")).json()
    assert body["subjects"] == {
        "tabular_request": app_module.settings.OPENBAO.NATS.NATS_TABULAR_REQUEST_SUBJECT,
        "image_request": app_module.settings.OPENBAO.NATS.NATS_IMAGE_REQUEST_SUBJECT,
        "result": app_module.settings.OPENBAO.NATS.NATS_RESULT_SUBJECT,
    }
    assert "NATS_JWT" not in str(body) and "jwt" not in str(body).lower()

и такой же test_healthcheck_subjects_when_nats_down с NATS_ENABLED=True без соединения.

tests/test_api_files_upload.py — тест создания задания с subject'ами окружения: подменить settings.OPENBAO.NATS на NATSSettings(..., NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded") через monkeypatch.setattr на объекте настроек, которым пользуется сервис загрузки в этом тест-файле, и проверить job.request_subject == "test.files.tabular_uploaded" (по образцу существующей проверки на строке ~251).

  • Step 2: тесты падают. Run: make -C $W/arm-core test и cd $W/arm-core/api && .venv/bin/pytest tests/test_recognition_exchange.py -q -k configured (нужен NATS стенда). Expected: FAIL (used_subjects нет, publish_pending не принимает request_subjects, нет subjects в healthcheck).

  • Step 3: реализация.

    src/core/nats.py:

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,
    }

В NatsConnection.__init__: result_subject: str | None = None, self.result_subject = result_subject or settings.NATS_RESULT_SUBJECT; импорт RESULT_SUBJECT из src.contract.subjects убрать, если больше не используется; docstring класса и модуля: «по умолчанию — NATS_RESULT_SUBJECT настроек».

src/services/recognition_exchange.py: _process_one(..., request_subjects: Mapping[str, str]) → subject = request_subject(row.pipeline.value, request_subjects); publish_pending(..., request_subjects: Mapping[str, str] = REQUEST_SUBJECTS) передаёт в _process_one; run_publisher_loop(..., request_subjects: Mapping[str, str] = REQUEST_SUBJECTS) передаёт в publish_pending; ExchangeTasks.start передаёт request_subjects=settings.OPENBAO.NATS.request_subjects(). REQUEST_SUBJECTS импортировать из src.contract.

src/services/primary_files.py:483: request_subject=request_subject(detected.pipeline.value, self._settings.OPENBAO.NATS.request_subjects()).

src/app.py: в lifespan перед _register_smb() — logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS)); в healthcheck — return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status, "subjects": used_subjects(app_settings.OPENBAO.NATS)}; docstring модуля: healthcheck (uptime, version, nats: ok|down|off, subjects).

.env.example: после OPENBAO__NATS__NATS_JWT= три строки с умолчаниями и комментарием # subject'ы обмена — по окружению (прод/тест), у трёх сервисов одинаковые; умолчания — имена контракта. README: в таблицу/список переменных NATS — три ключа; в описание /healthcheck — поле subjects.

  • Step 4: тесты проходят. Run: make -C $W/arm-core lint test test-integration Expected: PASS.

  • Step 5: Commit.

git -C $W/arm-core add api/src api/tests api/.env.example README.md
git -C $W/arm-core commit -m "NATS: публикация, подписка и /healthcheck.subjects по настройкам окружения (#106)"

Task 4: tabular-parser — настройки NATS, контракт, nkeys

Model: haiku (код в разделе A).

Files:

  • Modify: $W/tabular-parser/api/src/core/config.py (A.1–A.4)
  • Modify: $W/tabular-parser/api/src/contract/subjects.py (A.5)
  • Modify: $W/tabular-parser/api/src/core/nats.py — build_nats_auth (A.6)
  • Modify: $W/tabular-parser/api/contracts/recognition-request.schema.json, recognition-result-message.schema.json (копия из $W/docs/contracts/)
  • Test: $W/tabular-parser/api/tests/unit/test_config.py, $W/tabular-parser/api/tests/unit/test_contract.py (или файл, где сейчас тестируется request_subject; найти grep -rn request_subject tests)

Interfaces:

  • Consumes: Task 1.

  • Produces: то же, что Task 2, в tabular-parser.

  • Step 1: фабрика и тесты. В tests/unit/test_config.py фабрика settings_from_env поверх существующей фикстуры minimal_settings (conftest): если minimal_settings(**env) уже принимает переопределения — обёртка settings_from_env = minimal_settings с поддержкой None → monkeypatch.delenv; иначе — по образцу Task 2 Step 1 с обязательными для tabular-parser OPENBAO__MINIO__* и OPENBAO__NATS__NATS_URL. Тесты A.7 дословно; существующие NATS_SEED == ""/NATS_JWT == "" → .get_secret_value() == "". Тест request_subject(pipeline, subjects) — как в Task 2 Step 1 (test_request_subject_uses_environment_map).

  • Step 2: тесты падают. Run: make -C $W/tabular-parser test Expected: FAIL.

  • Step 3: реализация. A.1–A.6; cp $W/docs/contracts/recognition-request.schema.json $W/docs/contracts/recognition-result-message.schema.json $W/tabular-parser/api/contracts/.

  • Step 4: тесты проходят. Run: make -C $W/tabular-parser lint test Expected: PASS (тест идентичности контракта видит ../docs = $W/docs).

  • Step 5: Commit.

git -C $W/tabular-parser add api/src/core/config.py api/src/contract/subjects.py api/src/core/nats.py api/contracts api/tests
git -C $W/tabular-parser commit -m "NATS: subject'ы и JWT из OPENBAO__NATS__*, SecretStr, скрытие входа в ошибках (#31)"

Task 5: tabular-parser — подписка, публикация, лог, /healthcheck, roundtrip на своих subject'ах

Model: sonnet.

Files:

  • Modify: $W/tabular-parser/api/src/core/nats.py — subscribe_requests (умолчание subject'а), новая used_subjects
  • Modify: $W/tabular-parser/api/src/core/worker.py — RecognitionWorker.__init__ (result_subject), публикация (:467)
  • Modify: $W/tabular-parser/api/src/app.py — _build_worker, lifespan (лог), healthcheck, docstring модуля
  • Modify: $W/tabular-parser/api/.env.example, $W/tabular-parser/README.md
  • Test: $W/tabular-parser/api/tests/unit/test_nats.py, tests/unit/test_worker.py, tests/unit/test_app.py, tests/integration/test_roundtrip.py

Interfaces:

  • Consumes: Task 4.

  • Produces: used_subjects(settings: NATSSettings) -> dict[str, str] (ключи tabular_request, result); RecognitionWorker(..., result_subject: str = RESULT_SUBJECT); NatsConnection.subscribe_requests(cb, *, queue=REQUEST_QUEUE_GROUP, subject: str | None = None) — None → settings.NATS_TABULAR_REQUEST_SUBJECT.

  • Step 1: тесты.

    tests/unit/test_nats.py:

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"}

и тест, что subscribe_requests(cb) без subject подписывается на NATS_TABULAR_REQUEST_SUBJECT настроек — по образцу существующего unit-теста подписки в этом файле (фейковый клиент, проверка аргумента subscribe).

tests/unit/test_worker.py — по образцу существующего теста публикации ack/результата (фейковый publisher с записью (subject, payload, msg_id)): воркер с result_subject="test.files.tabular_parsed" публикует ack и результат в этот subject.

tests/unit/test_app.py — test_healthcheck_subjects_when_nats_off / _down по образцу существующих тестов healthcheck файла: body["subjects"] == {"tabular_request": ..., "result": ...} из app_module.settings, JWT в ответе нет.

tests/integration/test_roundtrip.py — уникальные subject'ы на прогон (OQ-83): в фикстуре service перед lifespan_context:

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",
}
# в фикстуре service:
monkeypatch.setattr(app_module.settings.OPENBAO, "NATS", app_module.settings.OPENBAO.NATS.model_copy(update=TEST_SUBJECTS))

(RUN и TEST_SUBJECTS — константы модуля); класс Core публикует в TEST_SUBJECTS["NATS_TABULAR_REQUEST_SUBJECT"], фикстура core подписывается на TEST_SUBJECTS["NATS_RESULT_SUBJECT"]; docstring модуля — убрать оговорку про общую шину, если она есть, добавить «subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83)».

  • Step 2: тесты падают. Run: make -C $W/tabular-parser test Expected: FAIL.

  • Step 3: реализация.

    src/core/nats.py:

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}

subscribe_requests: параметр subject: str | None = None, в теле subject = subject or self._settings.NATS_TABULAR_REQUEST_SUBJECT (имя атрибута настроек — как в классе); импорт TABULAR_REQUEST_SUBJECT убрать, если не используется; docstring модуля — «подписка на subject запроса tabular из настроек (по умолчанию files.tabular_uploaded)».

src/core/worker.py: RecognitionWorker.__init__(..., result_subject: str = RESULT_SUBJECT) → self._result_subject = result_subject; строка :467 — await self._publisher.publish(self._result_subject, payload, msg_id).

src/app.py: в _build_worker — RecognitionWorker(..., result_subject=settings.OPENBAO.NATS.NATS_RESULT_SUBJECT); в lifespan первым действием — logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS)); healthcheck — {"uptime": ..., "version": ..., "nats": ..., "subjects": used_subjects(settings.OPENBAO.NATS)}.

.env.example, README — как в Task 3 Step 3 (три ключа, поле subjects).

  • Step 4: тесты проходят. Run: make -C $W/tabular-parser lint test test-integration Expected: PASS (интеграционные — при поднятом compose arm-ka; контейнер tabular-parser стенда останавливать не нужно).

  • Step 5: Commit.

git -C $W/tabular-parser add api/src api/tests api/.env.example README.md
git -C $W/tabular-parser commit -m "NATS: подписка, публикация и /healthcheck.subjects по настройкам; roundtrip на своих subject'ах (#31)"

Task 6: scan-gateway — настройки NATS, контракт, nkeys

Model: haiku (код в разделе A).

Files:

  • Modify: $W/scan-gateway/api/src/core/config.py (A.1–A.4; у scan-gateway ещё SMBSettings, ServerSettings)
  • Modify: $W/scan-gateway/api/src/contract/subjects.py (A.5)
  • Modify: $W/scan-gateway/api/src/core/nats.py — build_nats_auth (A.6, сохранить .strip())
  • Modify: $W/scan-gateway/api/contracts/recognition-request.schema.json, recognition-result-message.schema.json
  • Test: $W/scan-gateway/api/tests/unit/test_config.py, tests/unit/test_contract.py, tests/unit/test_nats.py

Interfaces:

  • Consumes: Task 1.

  • Produces: то же, что Task 2, в scan-gateway.

  • Step 1: фабрика и тесты. В tests/unit/test_config.py фабрика settings_from_env поверх существующих _clean_env(monkeypatch) и фикстуры minimal_env_file (сначала _clean_env, затем setenv/delenv по env, затем get_settings.cache_clear() если get_settings кэширован, и Settings()); тесты A.7 дословно (для missing="OPENBAO__MINIO__MINIO_URL" переменная должна отсутствовать и в minimal_env_file — если она задаётся файлом, тест строит Settings(_env_file=None) с явным набором env). Существующие сравнения NATS_SEED == "" / NATS_JWT == "" → .get_secret_value() == "". Тест request_subject(pipeline, subjects) — как в Task 2.

  • Step 2: тесты падают. Run: make -C $W/scan-gateway test Expected: FAIL.

  • Step 3: реализация. A.1–A.6; копия схем в $W/scan-gateway/api/contracts/.

  • Step 4: тесты проходят. Run: make -C $W/scan-gateway lint test Expected: PASS.

  • Step 5: Commit.

git -C $W/scan-gateway add api/src/core/config.py api/src/contract/subjects.py api/src/core/nats.py api/contracts api/tests
git -C $W/scan-gateway commit -m "NATS: subject'ы и JWT из OPENBAO__NATS__*, SecretStr, скрытие входа в ошибках (#32)"

Task 7: scan-gateway — подписка, публикация, лог, /healthcheck, интеграционные тесты на своих subject'ах

Model: sonnet.

Files:

  • Modify: $W/scan-gateway/api/src/core/nats.py — NatsGateway.subscribe_requests (:211–237), publish_result (:294–298), новая used_subjects
  • Modify: $W/scan-gateway/api/src/core/health.py — snapshot
  • Modify: $W/scan-gateway/api/src/app.py — лог при старте, docstring модуля
  • Modify: $W/scan-gateway/api/.env.example, $W/scan-gateway/README.md
  • Test: $W/scan-gateway/api/tests/unit/test_nats.py, tests/unit/test_health.py, tests/unit/test_app.py, tests/integration/test_nats_minio_integration.py, tests/integration/test_roundtrip.py

Interfaces:

  • Consumes: Task 6.

  • Produces: used_subjects(settings: NATSSettings) -> dict[str, str] (ключи image_request, result); NatsGateway подписывается на settings.OPENBAO.NATS.NATS_IMAGE_REQUEST_SUBJECT и публикует в settings.OPENBAO.NATS.NATS_RESULT_SUBJECT.

  • Step 1: тесты.

    tests/unit/test_nats.py:

def test_used_subjects_scan_gateway() -> None:
    nats = NATSSettings(NATS_URL="nats://x:4222", NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded")
    assert used_subjects(nats) == {"image_request": "test.files.img_uploaded", "result": "files.tabular_parsed"}

и по образцу существующих unit-тестов NatsGateway с фейковым клиентом: при OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=test.files.img_uploaded и OPENBAO__NATS__NATS_RESULT_SUBJECT=test.files.tabular_parsed (make_settings(monkeypatch, ...)) subscribe_requests вызывает subscribe("test.files.img_uploaded", queue="scan-gateway", ...), а publish_result публикует в "test.files.tabular_parsed".

tests/unit/test_health.py — snapshot() содержит "subjects": {"image_request": ..., "result": ...} при nats_status ok, down и disabled.

tests/unit/test_app.py — /healthcheck отдаёт subjects, JWT в ответе нет.

Интеграционные (tests/integration/): по образцу Task 5 Step 1 — константы RUN, TEST_SUBJECTS и настройки тестовых NatsGateway с этими subject'ами (через make_settings/monkeypatch.setenv на OPENBAO__NATS__NATS_*_SUBJECT до создания Settings), публикация запросов и подписка на результат — в TEST_SUBJECTS; убрать фильтрацию «чужие сообщения на общем subject пропускаем», если после перехода она больше не нужна (оставить, если сообщения фильтруются по job_id внутри теста по другим причинам).

  • Step 2: тесты падают. Run: make -C $W/scan-gateway test Expected: FAIL.

  • Step 3: реализация.

    src/core/nats.py:

def used_subjects(settings: NATSSettings) -> dict[str, str]:
    """Subject'ы сервиса для лога старта и `/healthcheck.subjects`: подписан на запрос `image`, публикует ответ."""
    return {"image_request": settings.NATS_IMAGE_REQUEST_SUBJECT, "result": settings.NATS_RESULT_SUBJECT}

subscribe_requests: subject = self._settings.OPENBAO.NATS.NATS_IMAGE_REQUEST_SUBJECT вместо IMAGE_REQUEST_SUBJECT (в сообщении NatsNotConnectedError, в subscribe, в логе); publish_result: await self.publish(self._settings.OPENBAO.NATS.NATS_RESULT_SUBJECT, payload, ...); неиспользуемые импорты констант убрать; docstring'и — «subject из настроек (по умолчанию …)».

src/core/health.py: snapshot() возвращает {"nats": ..., "hotfolder": ..., "subjects": self._subjects}, где self._subjects — новый параметр __init__ subjects: Mapping[str, str] (передать в app.py при создании Health...: subjects=used_subjects(settings.OPENBAO.NATS)).

src/app.py: лог logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS)) в начале lifespan; docstring модуля — /healthcheck отдаёт uptime, version, nats, hotfolder, subjects.

.env.example, README — как в Task 3 Step 3.

  • Step 4: тесты проходят. Run: make -C $W/scan-gateway lint test test-integration Expected: PASS.

  • Step 5: Commit.

git -C $W/scan-gateway add api/src api/tests api/.env.example README.md
git -C $W/scan-gateway commit -m "NATS: подписка, публикация и /healthcheck.subjects по настройкам; интеграционные тесты на своих subject'ах (#32)"

Task 8: docs — документы контекста и окружения разработки

Model: sonnet.

Files:

  • Modify: $W/docs/context/03-architecture.md — схема/таблица потоков (строки ~17, ~68–76), раздел секретов (~306), раздел NATS платформы (~334)
  • Modify: $W/docs/context/07-open-questions.md — OQ-83 (строка ~91)
  • Modify: $W/docs/dev/README.md — таблица переменных (~323–325), раздел интеграционных тестов/e2e (~405), описание /healthcheck
  • Modify: $W/docs/dev/.env.example, $W/docs/dev/contour-transfer-guide.md (~180–188), $W/docs/dev/qa-guide.md (~955: /healthcheck), $W/docs/dev/frontend-api-guide.md (если описан ответ /healthcheck)
  • Modify: $W/docs/repo-templates/*.CLAUDE.md — только если там названы константы subject'ов как неизменные
  • Modify: $W/docs/openspec/changes/nats-env-config/tasks.md — отметить выполненные пункты

Interfaces:

  • Consumes: итог задач 1–7 (имена ключей, поле subjects, поведение тестов).

  • Step 1: 03-architecture.md. В таблице потоков и на схеме имена files.* оставить, пометив «subject'ы — настройки окружения OPENBAO__NATS__NATS_{TABULAR_REQUEST,IMAGE_REQUEST,RESULT}_SUBJECT, здесь — умолчания». В перечне секретов — три ключа рядом с NATS_URL/NATS_SEED/NATS_JWT для всех трёх сервисов. В разделе NATS платформы — абзац: «Прод и тест — свои наборы: JWT, seed и три subject'а окружения задаются одинаково всем трём сервисам; JWT окружения должен давать publish/subscribe на эти subject'ы. Разъехавшиеся настройки — тишина на шине (dlq по NO_ACK); сверять /healthcheck.subjects ядра и сервисов.» (решение владельца 12.10.2026).

  • Step 2: 07-open-questions.md, OQ-83. В колонку решения дописать: «13.10.2026 (nats-env-config): subject'ы — настройки; интеграционные тесты tabular-parser и scan-gateway идут на уникальных subject'ах прогона и не пересекаются со стендом. Для make e2e ядра вопрос остаётся: ему нужен контейнер tabular-parser стенда на умолчаниях, конфликт — общая queue group arm-core (design R-7)». Статус строки — открыт (частично).

  • Step 3: dev/README.md, dev/.env.example, contour-transfer-guide.md, qa-guide.md, frontend-api-guide.md. Три ключа в таблицах переменных (умолчание, смысл); в contour-transfer-guide.md — в перечень секретов каждого сервиса и пункт «для тестового контура — свой набор JWT/seed/subject'ов»; ответ /healthcheck — с полем subjects там, где он описан; в README — интеграционные тесты сервисов больше не требуют останавливать контейнеры стенда, make e2e ядра — по-прежнему требует (R-7).

  • Step 4: проверка. Run: make -C $W/docs check и cd $W/docs && openspec validate nats-env-config --strict. Expected: зелёный / valid.

  • Step 5: Commit.

git -C $W/docs add context dev repo-templates openspec/changes/nats-env-config/tasks.md
git -C $W/docs commit -m "docs: subject'ы и JWT NATS по окружению — архитектура, окружение, OQ-83 (#80)"

После задач (контроллер)

  • Отметить пункты tasks.md дочерних change'ей в трёх сервисах (каждый — коммитом в своей ветке).
  • make -C $W/<repo> check-full в трёх сервисах (без | tail: код возврата не терять). Для make e2e ядра — остановить ядро стенда на время прогона (R-7).
  • Финальное ревью ветки (opus) по четырём репозиториям, таблица находок с маркерами и статусами.