fix: nullable jsonb пишет SQL NULL вместо jsonb null (I-1)

Все nullable jsonb-колонки ORM (audit_event.before_value/after_value,
ex_snapshot_item.payload, recognition_job.group_schema/metrics,
osv_entry.original_values) объявлены через nullable_jsonb() =
JSONB(none_as_null=True): Python None сохраняется как SQL NULL, IS NULL
находит строку. DDL не меняется; сверка с DBML и миграции зелёные.

Обходы для не-объекта в metrics (CASE по jsonb_typeof в публикаторе,
merge_metrics в inbox) оставлены: строки до исправления и сырой SQL
могут содержать jsonb null. Тесты: SQL NULL после AuditService.record
без before/after, metrics фабрики None -> SQL NULL, повтор публикации и
слияние метрик для SQL NULL и jsonb null, перечень колонок с флагом.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api/src/models/audit_ex.py

@@ -10,10 +10,10 @@ import uuid
 from typing import Any
 
 from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Integer, SmallInteger, String, Text
-from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
-from .base import Base, UUIDPrimaryMixin
+from .base import Base, UUIDPrimaryMixin, nullable_jsonb
 from .enums import (
     AuditAction,
     AuditSection,
@@ -58,8 +58,8 @@ class AuditEvent(UUIDPrimaryMixin, Base):
         UUID(as_uuid=True), ForeignKey("consolidation_version.id")
     )
     description: Mapped[str | None] = mapped_column(Text)
-    before_value: Mapped[Any | None] = mapped_column(JSONB)
-    after_value: Mapped[Any | None] = mapped_column(JSONB)
+    before_value: Mapped[Any | None] = mapped_column(nullable_jsonb())
+    after_value: Mapped[Any | None] = mapped_column(nullable_jsonb())
     comment: Mapped[str | None] = mapped_column(Text)
     result: Mapped[str | None] = mapped_column(String)
     changed_field: Mapped[str | None] = mapped_column(String)
@@ -132,7 +132,7 @@ class ExSnapshotItem(UUIDPrimaryMixin, Base):
     item_type: Mapped[str] = mapped_column(String, nullable=False)
     legal_entity_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("legal_entity.id"))
     document_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("document.id"))
-    payload: Mapped[Any | None] = mapped_column(JSONB)
+    payload: Mapped[Any | None] = mapped_column(nullable_jsonb())
     payload_schema_version: Mapped[str | None] = mapped_column(String)
 
 

api/src/models/base.py

@@ -3,7 +3,7 @@
 import uuid
 
 from sqlalchemy import MetaData
-from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import DeclarativeBase, Mapped, declarative_mixin, mapped_column
 
 #: Схема PostgreSQL, в которой живёт вся модель данных (DBML v5.13).
@@ -21,6 +21,15 @@ NAMING_CONVENTION = {
 }
 
 
+def nullable_jsonb() -> JSONB:
+    """Тип nullable-колонки jsonb: Python `None` пишется как SQL NULL, а не jsonb `null` (`none_as_null=True`).
+
+    Без флага SQLAlchemy сохраняет `None` как jsonb `null`, и `IS NULL` в SQL такую строку не находит. DDL не
+    меняется (тип колонки — `jsonb`). Записать именно jsonb `null` можно явно значением `sqlalchemy.JSON.NULL`.
+    """
+    return JSONB(none_as_null=True)  # type: ignore[no-untyped-call]  # у JSONB.__init__ нет аннотаций
+
+
 class Base(DeclarativeBase):
     metadata = MetaData(schema=SCHEMA, naming_convention=NAMING_CONVENTION)
 

api/src/models/files.py

@@ -10,10 +10,10 @@ import uuid
 from typing import Any
 
 from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, SmallInteger, String, Text, text
-from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
-from .base import Base, UUIDPrimaryMixin
+from .base import Base, UUIDPrimaryMixin, nullable_jsonb
 from .enums import (
     DocumentCheckResult,
     DocumentStatus,
@@ -150,8 +150,8 @@ class RecognitionJob(UUIDPrimaryMixin, Base):
     error_message: Mapped[str | None] = mapped_column(Text)
     result_s3_object_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("s3_object.id"))
     result_schema_version: Mapped[str | None] = mapped_column(String)
-    group_schema: Mapped[Any | None] = mapped_column(JSONB)
-    metrics: Mapped[Any | None] = mapped_column(JSONB)
+    group_schema: Mapped[Any | None] = mapped_column(nullable_jsonb())
+    metrics: Mapped[Any | None] = mapped_column(nullable_jsonb())
     fields_total: Mapped[int | None] = mapped_column(Integer)
     fields_low_confidence: Mapped[int | None] = mapped_column(Integer)
     fields_unrecognized: Mapped[int | None] = mapped_column(Integer)

api/src/models/osv.py

@@ -10,10 +10,10 @@ import uuid
 from typing import Any
 
 from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, Numeric, SmallInteger, String, text
-from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
-from .base import Base, UUIDPrimaryMixin
+from .base import Base, UUIDPrimaryMixin, nullable_jsonb
 from .enums import OsvDataType, OsvGroupType, pg_enum
 
 
@@ -53,7 +53,7 @@ class OsvEntry(UUIDPrimaryMixin, Base):
     closing_debit: Mapped[decimal.Decimal | None] = mapped_column(Numeric(18, 2))
     closing_credit: Mapped[decimal.Decimal | None] = mapped_column(Numeric(18, 2))
     is_modified: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
-    original_values: Mapped[Any | None] = mapped_column(JSONB)
+    original_values: Mapped[Any | None] = mapped_column(nullable_jsonb())
     modified_by: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("app_user.id"))
     modified_at: Mapped[datetime.datetime | None] = mapped_column(DateTime(timezone=True))
     is_excisable: Mapped[bool | None] = mapped_column(Boolean)

api/src/services/recognition_exchange.py

@@ -276,7 +276,8 @@ async def _process_one(  # noqa: PLR0913 - один шаг прохода: вс
         )
     else:
         count = _republish_count(row.metrics) + 1
-        # jsonb `||`: прочие ключи metrics сохраняются; не объект (SQL NULL или JSON null — так ORM пишет None) → {}
+        # jsonb `||`: прочие ключи metrics сохраняются; не объект (SQL NULL или jsonb null — строки, записанные до
+        # JSONB(none_as_null=True) или сырым SQL) → {}
         is_object = func.jsonb_typeof(RecognitionJob.metrics) == "object"
         base = case((is_object, RecognitionJob.metrics), else_=literal({}, JSONB))
         metrics = base.op("||", return_type=JSONB)(literal({REPUBLISH_COUNT_KEY: count}, JSONB))
@@ -533,7 +534,8 @@ def header_msg_id(headers: Mapping[str, str] | None) -> str | None:
 def merge_metrics(existing: Any, incoming: Mapping[str, Any] | None) -> Any:
     """Метрики сервиса поверх метрик задания: ключи ядра (`republish_count`) сохраняются, совпадающие — от сервиса.
 
-    `incoming is None` — метрики задания как есть. Не объект (SQL NULL, JSON `null` — так ORM пишет `None`) — как `{}`.
+    `incoming is None` — метрики задания как есть. Не объект (SQL NULL; jsonb `null` в строках, записанных до
+    `JSONB(none_as_null=True)` или сырым SQL) — как `{}`.
     """
     if incoming is None:
         return existing

api/tests/factories.py

@@ -11,7 +11,7 @@ import decimal
 import uuid
 from typing import Any
 
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.models.directories import LegalEntity
@@ -171,6 +171,21 @@ class CommittedJobs:
             assert job is not None
             return job
 
+    async def metrics_sql_state(self, job_id: uuid.UUID) -> tuple[bool, str | None]:
+        """Как хранится `recognition_job.metrics` на уровне SQL: (`IS NULL`, `jsonb_typeof`).
+
+        SQL NULL — `(True, None)`; jsonb `null` — `(False, "null")`; объект — `(False, "object")`.
+        """
+        async with self._session_factory() as session:
+            row = (
+                await session.execute(
+                    select(RecognitionJob.metrics.is_(None), func.jsonb_typeof(RecognitionJob.metrics)).where(
+                        RecognitionJob.id == job_id
+                    )
+                )
+            ).one()
+            return bool(row[0]), row[1]
+
     def track_s3_object(self, s3_object_id: uuid.UUID) -> None:
         """Запомнить строку `s3_object`, созданную тестом вне фабрики (удаляется в `cleanup()`)."""
         self._extra_s3_ids.append(s3_object_id)

api/tests/test_audit_service.py

@@ -440,3 +440,17 @@ async def test_record_does_not_commit_and_external_rollback_discards(db_session:
 
     await db_session.rollback()  # внешний откат (в фикстуре — к SAVEPOINT) уничтожает запись
     assert await _count_events(db_session, object_id) == 0
+
+
+@pytest.mark.integration
+async def test_absent_before_after_stored_as_sql_null(db_session: AsyncSession) -> None:
+    """Без before/after в БД — SQL NULL, а не jsonb `null` (`JSONB(none_as_null=True)`, финальное ревью I-1)."""
+    event = await AuditService(db_session).record(
+        action=AuditAction.LOGIN, section=AuditSection.SESSION, object_type="app_user", object_id=uuid.uuid4()
+    )
+    is_null = (
+        await db_session.execute(
+            select(AuditEvent.before_value.is_(None), AuditEvent.after_value.is_(None)).where(AuditEvent.id == event.id)
+        )
+    ).one()
+    assert tuple(is_null) == (True, True)

api/tests/test_dbml_parity.py

@@ -91,6 +91,7 @@ def _dbml_enum(name: str) -> object:
         (Numeric(9, 4), "numeric(7,4)", False),
         (Integer(), "numeric(18,2)", False),
         (JSONB(), "jsonb", True),
+        (JSONB(none_as_null=True), "jsonb", True),  # SQL NULL вместо JSON null — DDL тот же
         (JSON(), "jsonb", False),
         (BigInteger(), "bigint", True),
         (Integer(), "bigint", False),

api/tests/test_models_part_07_09.py

@@ -3,7 +3,8 @@
 import datetime
 
 import pytest
-from sqlalchemy import inspect, text
+from sqlalchemy import Column, inspect, text
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
@@ -257,3 +258,30 @@ async def _check_active_lock_is_unique(session: AsyncSession) -> None:
         )
     ).one()
     assert tuple(counts) == (1, 1)
+
+
+NULLABLE_JSONB_COLUMNS = [
+    column
+    for table in Base.metadata.sorted_tables
+    for column in table.columns
+    if isinstance(column.type, JSONB) and column.nullable
+]
+
+
+def test_all_nullable_jsonb_columns_are_listed() -> None:
+    """Перечень nullable jsonb-колонок ORM (новая колонка попадёт под проверку ниже автоматически)."""
+    assert {f"{c.table.name}.{c.name}" for c in NULLABLE_JSONB_COLUMNS} == {
+        "audit_event.before_value",
+        "audit_event.after_value",
+        "ex_snapshot_item.payload",
+        "recognition_job.group_schema",
+        "recognition_job.metrics",
+        "osv_entry.original_values",
+    }
+
+
+@pytest.mark.parametrize("column", NULLABLE_JSONB_COLUMNS, ids=lambda c: f"{c.table.name}.{c.name}")
+def test_nullable_jsonb_writes_sql_null(column: Column[object]) -> None:
+    """Python `None` в nullable jsonb пишется как SQL NULL, а не jsonb `null` (финальное ревью I-1)."""
+    assert isinstance(column.type, JSONB)
+    assert column.type.none_as_null is True

api/tests/test_recognition_exchange.py

@@ -21,7 +21,7 @@ import pytest_asyncio
 from jsonschema import Draft202012Validator, FormatChecker
 from nats.aio.client import Client as NATS
 from nats.aio.msg import Msg
-from sqlalchemy import update
+from sqlalchemy import JSON, update
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 
 from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest
@@ -371,6 +371,33 @@ async def test_republish_without_ack_then_dlq(
     assert len(subscriber.for_job(job_id)) == 1 + MAX_REPUBLISH
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    ("stored", "expected_state"),
+    [
+        (None, (True, None)),  # Python None → SQL NULL (JSONB(none_as_null=True), финальное ревью I-1)
+        (JSON.NULL, (False, "null")),  # jsonb null: строки, записанные до исправления или сырым SQL
+    ],
+    ids=["sql_null", "json_null"],
+)
+async def test_republish_counter_on_null_metrics(
+    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, stored: Any, expected_state: Any
+):
+    """Обход для не-объекта в `metrics` (`jsonb_typeof` CASE) работает и для SQL NULL, и для jsonb null."""
+    clock = Clock()
+    job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics=stored)
+    assert await jobs.metrics_sql_state(job_id) == expected_state
+    fake = FakeNats()
+
+    assert (await _publish(session_factory, fake, clock)).published == 1
+    assert await jobs.metrics_sql_state(job_id) == expected_state  # первая публикация metrics не трогает
+    clock.advance(ACK + timedelta(seconds=1))
+    assert (await _publish(session_factory, fake, clock)).republished == 1
+
+    assert (await jobs.get(job_id)).metrics == {"republish_count": 1}
+    assert await jobs.metrics_sql_state(job_id) == (False, "object")
+
+
 @pytest.mark.integration
 async def test_ack_after_first_republish_stops_republishing(
     session_factory: async_sessionmaker[AsyncSession],

api/tests/test_recognition_inbox.py

@@ -26,7 +26,7 @@ from jsonschema import Draft202012Validator, FormatChecker
 from loguru import logger
 from nats.aio.client import Client as NATS
 from nats.aio.msg import Msg
-from sqlalchemy import func, select
+from sqlalchemy import JSON, func, select
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
 
 from src.contract import NATS_MSG_ID_HEADER, RecognitionResultMessage, ack_msg_id, result_msg_id
@@ -184,7 +184,7 @@ def test_header_msg_id_is_case_insensitive(headers: dict[str, str] | None, expec
     [
         ({"republish_count": 2}, {"duration_ms": 5}, {"republish_count": 2, "duration_ms": 5}),
         ({"republish_count": 2, "rows": 1}, {"rows": 7}, {"republish_count": 2, "rows": 7}),
-        (None, {"duration_ms": 5}, {"duration_ms": 5}),  # SQL NULL / JSON null (так ORM пишет None)
+        (None, {"duration_ms": 5}, {"duration_ms": 5}),  # SQL NULL / jsonb null (старые строки, сырой SQL)
         ([None, {"x": 1}], {"duration_ms": 5}, {"duration_ms": 5}),  # не объект — как пустой
         ({"republish_count": 1}, None, {"republish_count": 1}),  # метрик сервиса нет — без изменений
         (None, None, None),
@@ -377,6 +377,37 @@ async def test_needs_check_result(session_factory, jobs: CommittedJobs):
     assert job["metrics"] is None  # метрик не было ни у задания, ни у сервиса: колонка не трогается
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    ("stored", "expected_state"),
+    [(None, (True, None)), (JSON.NULL, (False, "null"))],
+    ids=["sql_null", "json_null"],
+)
+async def test_result_metrics_merge_over_null_metrics(
+    session_factory, jobs: CommittedJobs, stored: Any, expected_state: Any
+):
+    """Слияние метрик в inbox (Python `merge_metrics`) работает и для SQL NULL, и для jsonb null (финальное ревью I-1);
+    результат без метрик колонку не трогает: SQL NULL остаётся SQL NULL."""
+    with_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
+    without_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
+    assert await jobs.metrics_sql_state(with_metrics) == expected_state
+
+    file_id = await _file_id(jobs, with_metrics)
+    body = envelope(with_metrics, file_id, "done", metrics={"duration_ms": 10})
+    await _handle(session_factory, fake_msg(body, result_msg_id(with_metrics, 1)))
+    file_id = await _file_id(jobs, without_metrics)
+    await _handle(
+        session_factory, fake_msg(envelope(without_metrics, file_id, "done"), result_msg_id(without_metrics, 1))
+    )
+
+    job = await _row(session_factory, with_metrics)
+    assert job["status"] is RecognitionStatus.done
+    assert job["metrics"] == {"duration_ms": 10}
+    assert await jobs.metrics_sql_state(with_metrics) == (False, "object")
+    assert (await _row(session_factory, without_metrics))["status"] is RecognitionStatus.done
+    assert await jobs.metrics_sql_state(without_metrics) == expected_state
+
+
 @pytest.mark.integration
 async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
     """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object не меняются; mapper не вызывается."""